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

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

该示例是一个“汇率智能体（Currency Agent）”，通过 LangChain 与 LangGraph 调用 Gemini 模型来回答货币兑换问题。

准备 LangGraph 示例

1. 获取 API Key：如果还没有，请先创建一个 Gemini API Key 。

2. 配置环境变量：在 a2a-samples/samples/python/agents/langgraph/ 目录下创建 .env 文件：

echo "GOOGLE_API_KEY=YOUR_API_KEY_HERE" > .env

把 YOUR_API_KEY_HERE 替换为你自己的 Gemini API Key。请不要把真实的密钥提交到代码仓库。

3. 安装依赖：langgraph 示例有自己的 pyproject.toml，其中声明了 langchain-google-genai、langgraph 等依赖。如果运行时出现导入错误，请按该文件安装缺失的依赖。

在终端中进入 a2a-samples/samples/python/agents/langgraph/app 目录，并确认已激活虚拟环境。

启动 LangGraph 智能体服务端：

python __main__.py

服务端通常监听 http://localhost:10000。

打开一个新的终端窗口，激活虚拟环境，并进入 a2a-samples/samples/python/agents/langgraph/app，运行它的测试客户端：

python test_client.py

完成后，在运行 __main__.py 的终端中按 Ctrl+C 关闭服务端。

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

1. LLM 集成

agent.py 定义了 CurrencyAgent，使用 ChatGoogleGenerativeAI 与 LangGraph 的 create_react_agent 处理用户查询；
这说明真实的 LLM 如何驱动智能体的逻辑。

2. 任务状态管理

samples/python/agents/langgraph/app/__main__.py 用 InMemoryTaskStore 初始化 DefaultRequestHandler；
CurrencyAgentExecutor 的 execute 方法被 DefaultRequestHandler 调用时，会与包含当前任务（如果有）的 RequestContext 交互；
对于 SendMessage，DefaultRequestHandler 用 TaskStore 在多次交互之间持久化并取回任务状态。如果智能体的执行流程涉及多个步骤，或产生了持久化的任务，SendMessage 的响应就是完整的 Task 对象；
test_client.py 的 run_single_turn_test 演示了取得 Task 对象并用 get_task 查询它。

3. 通过 TaskStatusUpdateEvent 与 TaskArtifactUpdateEvent 流式传输

CurrencyAgentExecutor 的 execute 方法负责处理非流式与流式请求，并由 DefaultRequestHandler 统一编排；
当 LangGraph 智能体处理请求（可能调用 get_exchange_rate 等工具）时，CurrencyAgentExecutor 向 EventQueue 入队不同类型的事件：
- TaskStatusUpdateEvent：用于中间状态更新（例如“正在查询汇率…”“正在处理汇率…”）；
- TaskArtifactUpdateEvent：最终答案就绪时作为产出物入队，lastChunk 标志为 True；
- 最后发送一个 state=TaskState.TASK_STATE_COMPLETED 的 TaskStatusUpdateEvent，标志任务结束并关闭流；
test_client.py 的 run_streaming_test 函数会打印从服务端收到的每个事件数据块。

4. 多轮对话（TaskState.TASK_STATE_INPUT_REQUIRED）

当查询含糊不清时（例如用户问“100 美元是多少？”），CurrencyAgent 可以请求澄清；
此时 CurrencyAgentExecutor 会入队一个 TaskStatusUpdateEvent，其 status.state 为 TaskState.TASK_STATE_INPUT_REQUIRED，status.message 中是智能体的问题（例如“您想兑换成哪种货币？”），该事件之后流关闭；
test_client.py 的 run_multi_turn_test 函数演示了这个过程：
- 先发送一个含糊的查询；
- 智能体（由 DefaultRequestHandler 处理入队的事件）返回状态为 TASK_STATE_INPUT_REQUIRED 的 Task；
- 客户端再发送第二条消息，带上第一轮 Task 响应中的 taskId 与 contextId，补充缺失的信息（“换成英镑”），继续同一个任务。

这些状态的含义可参考任务的生命周期，流式传输与推送通知的细节见流式传输与异步操作。

建议花点时间阅读这些文件：

__main__.py：使用 DefaultRequestHandler 与 Starlette 应用搭建服务端。注意 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 如何支持智能体之间复杂、有状态的异步交互。