使用 A2A 协议和 LangGraph 构建你的第一个 AI 聊天机器人

mkdir a2a-langgraph-demo
cd a2a-langgraph-demo
uv venv
uv init

[project]
name = "a2a-langgraph-demo"
version = "0.1.0"
dependencies = [
    "a2a-sdk[http-server]>=0.3",
    "langgraph>=0.2",
    "langchain-core>=0.3",
    "langchain-openai>=0.2",
    "uvicorn[standard]>=0.30",
    "python-dotenv>=1.0.1"
]

from typing import Annotated
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict

# 定义状态：使用 add_messages 确保消息列表是追加而非覆盖
class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def model_node(state: ChatState):
    # 建议在 .env 中配置 OPENAI_BASE_URL 为 n1n.ai 的地址
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def build_app():
    workflow = StateGraph(ChatState)
    workflow.add_node("chatbot", model_node)
    workflow.add_edge(START, "chatbot")
    workflow.add_edge("chatbot", END)
    return workflow.compile()

chatbot_app = build_app()

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message
from langchain_core.messages import HumanMessage
from agent import chatbot_app

class MyAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 从 A2A 上下文中获取用户输入
        user_text = context.get_user_input()

        if not user_text:
            await event_queue.enqueue_event(new_agent_text_message("请输入有效内容。"))
            return

        # 调用 LangGraph 流程
        # 注意：在生产环境建议使用异步 ainvoke
        inputs = {"messages": [HumanMessage(content=user_text)]}
        result = await chatbot_app.ainvoke(inputs)

        # 提取最后一条 AI 回复
        ai_reply = result["messages"][-1].content

        # 将结果推送到 A2A 事件队列
        await event_queue.enqueue_event(new_agent_text_message(ai_reply))

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 暂不支持取消操作
        raise NotImplementedError("Cancel not supported.")

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
from agent_executor import MyAgentExecutor
from dotenv import load_dotenv

load_dotenv()

def start_server():
    # 定义技能
    chat_skill = AgentSkill(
        id="chat",
        name="基础聊天",
        description="使用 LangGraph 驱动的标准化聊天机器人",
        examples=["你好", "请介绍一下 A2A 协议"]
    )

    # 定义智能体卡片
    card = AgentCard(
        name="LangGraph-A2A-Bot",
        description="基于 A2A 协议的示例智能体",
        url="http://localhost:9999/",
        capabilities=AgentCapabilities(streaming=True),
        skills=[chat_skill]
    )

    # 绑定执行器
    handler = DefaultRequestHandler(
        agent_executor=MyAgentExecutor(),
        task_store=InMemoryTaskStore()
    )

    # 构建应用
    a2a_app = A2AStarletteApplication(agent_card=card, http_handler=handler)
    app = a2a_app.build()

    uvicorn.run(app, host="0.0.0.0", port=9999)

if __name__ == "__main__":
    start_server()