从零搭建 AI Agent 应用

pip install langchain langchain-anthropic langgraph tavily-python

from langchain.tools import tool
from datetime import datetime

@tool
def get_current_time() -> str:
    """获取当前时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def search_web(query: str) -> str:
    """搜索网页获取实时信息"""
    from tavily import TavilyClient
    client = TavilyClient()
    results = client.search(query, max_results=3)
    return "\n".join([r["content"] for r in results["results"]])

@tool
def execute_python(code: str) -> str:
    """执行 Python 代码并返回结果"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout or result.stderr

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import AnyMessage, add_messages
from langchain_anthropic import ChatAnthropic

# 1. 严格定义 Agent 的全局状态 (State Schema)
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    current_task: str

# 2. 节点逻辑：模型推理节点
def call_model(state: AgentState):
    llm = ChatAnthropic(model="claude-sonnet-5-5")
    tools = [get_current_time, search_web, execute_python]
    llm_with_tools = llm.bind_tools(tools)
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 3. 构建有向图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools)) # 内置的工具执行节点

# 4. 定义路由与边 (Edges)
workflow.set_entry_point("agent")
# 如果模型回复包含 tool_calls，流向 tools 节点，否则结束
workflow.add_conditional_edges("agent", lambda x: "tools" if x["messages"][-1].tool_calls else END)
workflow.add_edge("tools", "agent")

# 5. 编译成可执行应用
app = workflow.compile()

inputs = {"messages": [("user", "帮我查一下大模型的最新发布时间。")]}
for event in app.stream(inputs, stream_mode="values"):
    message = event["messages"][-1]
    message.pretty_print() # 能够清晰看到 Thought -> Action -> Observation 的每一步断点

from langgraph.checkpoint.redis import RedisSaver
import redis

# 建立底层 Redis 连接池
redis_conn = redis.Redis(host='localhost', port=6379, db=0)

# 使用持久化编译 Graph
with RedisSaver(redis_conn) as checkpointer:
    app = workflow.compile(
        checkpointer=checkpointer,
        interrupt_before=["tools"] # 在调用核心工具前实现硬中断，等待人工二次确认
    )
    
    # 通过 thread_id 区分不同用户会话，实现绝对的并发状态隔离
    config = {"configurable": {"thread_id": "user_10086_task_v2"}}
    
    # Agent 执行到 tools 节点前会自动挂起，状态安全保存在 Redis 中
    for event in app.stream(inputs, config=config):
        print(event)