多 Agent 框架集成

LangGraph Agent 将任务发布到 Janus 邮箱，并从下游 Agent 拉取结果。Janus 节点充当图步骤之间可靠的交接点：

from langgraph.graph import StateGraph
from janus_broker import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class AgentState(TypedDict):
    pr_url: str
    review_result: dict

def review_node(state: AgentState):
    client.publish_task({
        "source_agent": "langgraph-review",
        "target_type": "mailbox",
        "target_value": "review-mb",
        "envelope": {"type": "review", "payload": state}
    })
    return state

def collect_node(state: AgentState):
    result = client.pull_task("langgraph-collect-mb", "collector")
    return {"review_result": result.task.envelope.payload}

graph = StateGraph(AgentState)
graph.add_node(review_node)
graph.add_node(collect_node)
graph.add_edge("review_node", "collect_node")

LangGraph 负责编排图；Janus 确保任务在 Agent 重启后依然存活，并在步骤之间提供治理。

AutoGen Agent 将 Janus 用作异步团队协作的消息总线：

from autogen import AssistantAgent, UserProxyAgent
from janus_broker import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class JanusAgent(AssistantAgent):
    """通过 Janus 收发消息的 AutoGen Agent。"""

    def send(self, message, recipient, **kwargs):
        client.publish_task({
            "source_agent": self.name,
            "target_type": "mailbox",
            "target_value": f"{recipient}-mb",
            "envelope": {"type": "autogen_msg", "payload": message}
        })

    def receive(self, message, sender, **kwargs):
        # 消息经由持久化邮箱到达
        # Agent 就绪时拉取
        result = client.pull_task(f"{self.name}-mb", self.name)
        return result.task.envelope.payload

# 基于 Janus 骨干的 AutoGen 团队
reviewer = JanusAgent(name="reviewer", llm_config=llm_config)
coder = JanusAgent(name="coder", llm_config=llm_config)

CrewAI Agent 使用 Janus 在团队成员之间持久化交接任务：

from crewai import Agent, Task, Crew
from janus_broker import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class JanusCrewAgent(Agent):
    """带 Janus 任务队列的 CrewAI Agent。"""

    def execute_task(self, task, context=None):
        # 向 Janus 发布任务，交给下一位成员
        client.publish_task({
            "source_agent": self.role,
            "target_type": "mailbox",
            "target_value": f"{task.agent.role}-mb",
            "envelope": {"type": "crewai_task", "payload": task.description}
        })
        # 从自己的邮箱拉取结果
        result = client.pull_task(f"{self.role}-mb", self.role)
        return result.task.envelope.payload

# 具备 Janus 弹性的 Crew
crew = Crew(
    agents=[reviewer, coder, tester],
    tasks=[review_task, code_task, test_task],
    process=Process.sequential
)

GitHub Actions 触发 Janus 工作流并等待 Agent 结果：

# .github/workflows/agent-review.yml
name: Agent Code Review
on: [pull_request]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Request AI review via Janus
        run: |
          pip install janus-sdk
          python -c "
          from janus_broker import JanusClient
          client = JanusClient('${{ secrets.JANUS_URL }}',
                              api_key='${{ secrets.JANUS_API_KEY }}')
          client.publish_task({
              'source_agent': 'github-actions',
              'target_type': 'mailbox',
              'target_value': 'review-mb',
              'envelope': {
                  'type': 'pr_review',
                  'payload': {'pr': '${{ github.event.pull_request.html_url }}'}
              }
          })
          "

      - name: Wait for review result
        run: |
          python -c "
          import time
          from janus_broker import JanusClient
          client = JanusClient('${{ secrets.JANUS_URL }}',
                              api_key='${{ secrets.JANUS_API_KEY }}')
          while True:
              result = client.pull_task('github-results-mb', 'github-actions')
              if result:
                  print('Review complete:', result.task.envelope.payload)
                  break
              time.sleep(5)
          "