AI Agent 架构设计：从 ReAct 到 Multi-Agent 系统

Thought → Action → Observation → Thought → Action → ... → Final Answer

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next: str

def agent(state: AgentState):
    # LLM 推理：决定下一步行动
    response = llm.invoke([
        SystemMessage(content="你是一个能使用工具的 AI 助手"),
        *state["messages"]
    ])
    return {"messages": [response]}

def tools(state: AgentState):
    # 执行工具调用
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls"):
        tool_results = execute_tools(last_message.tool_calls)
        return {"messages": tool_results}
    return {"next": "end"}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent)
workflow.add_node("tools", tools)

workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "tools", "end": END}
)
workflow.add_edge("tools", "agent")

app = workflow.compile()

         ┌─────────────┐
         │  Supervisor  │  ← 任务分配与协调
         │   Agent      │
         └──┬──────┬───┘
            │      │
     ┌──────▼─┐ ┌──▼──────┐
     │  Coder │ │Researcher│
     │  Agent │ │  Agent   │
     └────────┘ └─────────┘

from langgraph.graph import MessageGraph

# 创建专家 Agent
researcher = create_agent("研究者", "搜索和分析信息")
coder = create_agent("程序员", "编写和调试代码")
reviewer = create_agent("审阅者", "检查和优化输出")

graph = MessageGraph()
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)

graph.add_edge("researcher", "coder")
graph.add_edge("coder", "reviewer")
graph.add_edge("reviewer", "researcher")

tools = [
    Tool(
        name="search",
        func=web_search,
        description="搜索互联网获取最新信息"
    ),
    Tool(
        name="calculator",
        func=calculator,
        description="执行数学计算"
    ),
    Tool(
        name="file_reader",
        func=read_file,
        description="读取本地文件内容"
    ),
    Tool(
        name="code_executor",
        func=execute_code,
        description="执行 Python 代码并返回结果"
    ),
]

class AgentMemory:
    def __init__(self):
        self.short_term = []  # 对话历史
        self.long_term = {}   # 向量化存储

    def remember(self, content, importance):
        """记住重要信息"""
        if importance > 0.7:
            embedding = get_embedding(content)
            self.long_term[content] = embedding

    def recall(self, query):
        """检索相关记忆"""
        query_emb = get_embedding(query)
        return semantic_search(query_emb, self.long_term)