AI Agent 生产落地的技术破局：基于 LangGraph 构建可控、可观测的多智能体系统

while max_iterations > 0:
    response = llm.invoke(messages, tools=tools)
    if response.tool_calls:
        execute_tools(response.tool_calls)
        messages.append(response)
    else:
        return response.content

from typing import Annotated, List, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import add_messages

class AgentState(TypedDict):
    # 消息列表使用了 add_messages 归约器，自动追加而非覆盖
    messages: Annotated[Sequence[BaseMessage], add_messages]
    # 当前规划的任务队列
    task_queue: List[dict]
    # 工具执行记录（用于观测）
    tool_execution_logs: List[dict]
    # 重试计数器（防止死循环）
    retry_count: int
    # 最终是否完成
    is_complete: bool

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 持久化状态
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    graph = builder.compile(checkpointer=saper)
    
    # 第一次运行，指定 thread_id
    config = {"configurable": {"thread_id": "user_123_session_456"}}
    for event in graph.stream(initial_state, config=config):
        print(event)
    
    # 若任务中断或需要恢复，直接传入相同 thread_id 继续执行
    next_state = graph.get_state(config)
    if not next_state.values.get("is_complete"):
        # 从断点处恢复
        for event in graph.stream(None, config=config):
            ...

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class ExecuteSQLInput(BaseModel):
    query: str = Field(description="标准SQL查询语句")
    database: str = Field(description="目标数据库名称")
    limit: int = Field(default=100, ge=1, le=10000)

@tool("execute_sql", args_schema=ExecuteSQLInput, return_direct=False)
def execute_sql(query: str, database: str, limit: int = 100) -> str:
    """执行只读SQL查询，返回JSON结果"""
    # 安全检查：强制添加 LIMIT，禁止 DDL/DML
    if not query.strip().upper().startswith("SELECT"):
        raise ValueError("只允许SELECT查询")
    # ... 执行逻辑
    return results_json

def parallel_executor_node(state: AgentState):
    tasks = state["task_queue"]
    # 使用 asyncio.gather 并发执行
    import asyncio
    loop = asyncio.new_event_loop()
    results = loop.run_until_complete(
        asyncio.gather(*[execute_single_task(t) for t in tasks])
    )
    # 聚合结果到 messages
    return {"messages": [ToolMessage(content=json.dumps(results))], "task_queue": []}

graph.compile(checkpointer=saver)
# 执行时强制最大递归深度
config.update({"recursion_limit": 25})

def router_after_validator(state: AgentState):
    if state["retry_count"] > 3:
        return "fallback"  # 进入兜底节点，直接告知用户超时
    if state["is_complete"]:
        return "end"
    return "planner"  # 继续规划

from langchain_core.callbacks import BaseCallbackHandler

class MetricCallbackHandler(BaseCallbackHandler):
    def on_tool_start(self, serialized, input_str, **kwargs):
        self.start_time = time.time()
        print(f"[Trace] 工具调用开始: {serialized.get('name')}")
    
    def on_tool_end(self, output, **kwargs):
        duration = time.time() - self.start_time
        # 推送到 Prometheus Counter/Histogram
        tool_duration_histogram.labels(name=self.tool_name).observe(duration)
        # 记录昂贵的Token消耗
        token_counter.add(llm_usage.total_tokens)

# 在 invoke 时传入
graph.invoke(initial_state, config={"callbacks": [MetricCallbackHandler()]})

# 错误写法
state["my_dict"]["key"] = "new" 
return {"my_dict": state["my_dict"]} # 可能不触发更新

# 正确写法
new_dict = state["my_dict"].copy()
new_dict["key"] = "new"
return {"my_dict": new_dict}

async def stream_agent(user_input: str):
    config = {"configurable": {"thread_id": user_id}}
    async for event in graph.astream_events(
        {"messages": [HumanMessage(content=user_input)]},
        config=config,
        version="v2"
    ):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            # 流式吐出 LLM 生成的文本
            yield f"data: {json.dumps({'type': 'token', 'content': event['data']['chunk'].content})}\n\n"
        elif kind == "on_tool_start":
            # 通知前端正在调用工具，展示加载状态
            yield f"data: {json.dumps({'type': 'tool_start', 'tool': event['name']})}\n\n"
        elif kind == "on_tool_end":
            yield f"data: {json.dumps({'type': 'tool_end', 'result': event['data']['output']})}\n\n"