7. 流式传输与多轮交互（LangGraph 示例）¶

Hello World 示例演示了 A2A 的基本机制。对于更高级的功能，例如健壮的流式传输、任务状态管理以及由 LLM 驱动的多轮对话，我们将转向位于 a2a-samples/samples/python/agents/langgraph/ 的 LangGraph 示例。

此示例包含一个“货币代理”，它通过 LangChain 和 LangGraph 使用 Gemini 模型来回答货币换算问题。

设置 LangGraph 示例¶

如果您还没有 Gemini API 密钥，请创建一个。
环境变量

在 a2a-samples/samples/python/agents/langgraph/ 目录中创建一个 .env 文件
```
echo "GOOGLE_API_KEY=YOUR_API_KEY_HERE" > .env
```
将 YOUR_API_KEY_HERE 替换为您的实际 Gemini API 密钥。
安装依赖项（如果尚未涵盖）

langgraph 示例有自己的 pyproject.toml，其中包含 langchain-google-genai 和 langgraph 等依赖项。当您从 a2a-samples 根目录使用 pip install -e .[dev] 安装 SDK 时，这也应该安装工作区示例的依赖项，包括 langgraph-example。如果您遇到导入错误，请确保从根目录成功安装了主 SDK。

运行 LangGraph 服务器¶

在终端中导航到 a2a-samples/samples/python/agents/langgraph/app 目录，并确保您的虚拟环境（来自 SDK 根目录）已激活。

启动 LangGraph 代理服务器

python __main__.py

这将启动服务器，通常在 https://:10000 上。

与 LangGraph 代理交互¶

打开一个新的终端窗口，激活您的虚拟环境，并导航到 a2a-samples/samples/python/agents/langgraph/app。

运行其测试客户端

python test_client.py

现在，您可以通过在运行 __main__.py 的终端窗口中键入 Ctrl+C 来关闭服务器。

演示的主要功能¶

langgraph 示例展示了几个重要的 A2A 概念

LLM 集成:
- agent.py 定义了 CurrencyAgent。它使用 ChatGoogleGenerativeAI 和 LangGraph 的 create_react_agent 来处理用户查询。
- 这演示了真实的 LLM 如何为代理的逻辑提供支持。
任务状态管理:
- samples/langgraph/__main__.py 使用 InMemoryTaskStore 初始化 DefaultRequestHandler。
```
httpx_client = httpx.AsyncClient()
push_config_store = InMemoryPushNotificationConfigStore()
push_sender = BasePushNotificationSender(httpx_client=httpx_client,
                config_store=push_config_store)
request_handler = DefaultRequestHandler(
    agent_executor=CurrencyAgentExecutor(),
    task_store=InMemoryTaskStore(),
    push_config_store=push_config_store,
    push_sender= push_sender
)
server = A2AStarletteApplication(
    agent_card=agent_card, http_handler=request_handler
)

uvicorn.run(server.build(), host=host, port=port)
```
- 当 DefaultRequestHandler 调用 CurrencyAgentExecutor（在 samples/langgraph/agent_executor.py 中）的 execute 方法时，它与包含当前任务（如果有）的 RequestContext 交互。
- 对于 message/send，DefaultRequestHandler 使用 TaskStore 在交互过程中持久化和检索任务状态。如果代理的执行流涉及多个步骤或导致持久任务，则 message/send 的响应将是一个完整的 Task 对象。
- test_client.py 的 run_single_turn_test 演示了获取 Task 对象，然后使用 get_task 查询它。
使用 TaskStatusUpdateEvent 和 TaskArtifactUpdateEvent 进行流式传输:
- CurrencyAgentExecutor 中的 execute 方法负责处理非流式和流式请求，由 DefaultRequestHandler 进行协调。
- 当 LangGraph 代理处理请求时（可能涉及调用 get_exchange_rate 等工具），CurrencyAgentExecutor 会将不同类型的事件排入 EventQueue
  - TaskStatusUpdateEvent：用于中间更新（例如，“正在查询汇率...”，“正在处理汇率...”）。这些事件上的 final 标志为 False。
  - TaskArtifactUpdateEvent：当最终答案准备好时，它会作为工件排队。lastChunk 标志为 True。
  - 发送一个带有 state=TaskState.completed 和 final=True 的最终 TaskStatusUpdateEvent，以表示流式传输任务的结束。
- test_client.py 的 run_streaming_test 函数将打印从服务器接收到的这些单独的事件块。
多轮对话 (TaskState.input_required):
- 如果查询不明确（例如，用户询问“100 美元是多少？”），CurrencyAgent 可以要求澄清。
- 发生这种情况时，CurrencyAgentExecutor 将将 TaskStatusUpdateEvent 入队，其中 status.state 为 TaskState.input_required，status.message 包含代理的问题（例如，“您想转换为哪种货币？”）。此事件将对当前交互流具有 final=True。
- test_client.py 的 run_multi_turn_test 函数演示了这一点
  - 它发送一个初始的模糊查询。
  - 代理（通过 DefaultRequestHandler 处理入队事件）以一个状态为 input_required 的 Task 响应。
  - 然后客户端发送第二条消息，其中包括来自第一轮 Task 响应的 taskId 和 contextId，以提供缺失的信息（“in GBP”）。这会继续相同的任务。

探索代码¶

花点时间查看这些文件

__main__.py：使用 A2AStarletteApplication 和 DefaultRequestHandler 进行服务器设置。请注意 AgentCard 定义包含 capabilities.streaming=True。
agent.py：带有 LangGraph、LLM 模型和工具定义的 CurrencyAgent。
agent_executor.py：实现 execute（和 cancel）方法的 CurrencyAgentExecutor。它使用 RequestContext 来理解正在进行的任务，并使用 EventQueue 来发送各种事件（TaskStatusUpdateEvent、TaskArtifactUpdateEvent，如果不存在任务，则隐式通过第一个事件发送新的 Task 对象）。
test_client.py：演示各种交互模式，包括检索任务 ID 并将其用于多轮对话。

此示例更丰富地说明了 A2A 如何促进代理之间复杂、有状态和异步的交互。