随着AI Agent技术进入主流应用, 掌握Agent框架成为AI工程师必备技能。本教程带你使用Python和LangGraph构建一个具备Web搜索和计算能力的AI助手。

环境准备

pip install langgraph langchain-openai langchain-community tavily-python

import os
os.environ["OPENAI_API_KEY"] = "your-key"
os.environ["TAVILY_API_KEY"] = "your-key"

第一步: 定义State

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]
    next_step: str

第二步: 构建工具函数

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool

@tool
def web_search(query: str) -> str:
    search = TavilySearchResults(max_results=3)
    results = search.invoke(query)
    return "\n".join([r["content"][:500] for r in results])

@tool
def calculate(expression: str) -> str:
    try:
        return f"Result: {eval(expression)}"
    except Exception as e:
        return f"Error: {str(e)}"

tools = [web_search, calculate]

第三步: 构建LangGraph图

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    last_msg = state["messages"][-1]
    for tc in last_msg.tool_calls:
        if tc["name"] == "web_search":
            result = web_search.invoke(tc["args"])
        elif tc["name"] == "calculate":
            result = calculate.invoke(tc["args"])
    return {"messages": [AIMessage(content=str(result))]}

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tool", tool_node)
workflow.set_entry_point("agent")

def should_continue(state):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tool"
    return "end"

workflow.add_conditional_edges("agent", should_continue, {"tool": "tool", "end": END})
workflow.add_edge("tool", "agent")
app = workflow.compile()

第四步: 运行Agent

result = app.invoke({
    "messages": [
        SystemMessage(content="You are a helpful AI assistant."),
        HumanMessage(content="Search latest AI funding news for May 2026")
    ]
})
print(result["messages"][-1].content)

进阶: 添加记忆功能

from langgraph.checkpoint import MemorySaver
memory = MemorySaver()
app_with_memory = workflow.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "session_1"}}

最佳实践

1. 安全优先: 设置工具权限边界, 参考微软AI Agent安全工具
2. 可观测性: 使用LangSmith记录推理过程
3. 降级策略: 工具调用失败时设计优雅降级
4. 成本控制: 设置最大推理步数防无限循环

LangGraph v0.5+已支持分布式执行和子Agent模式。掌握这些基础将为AI工程发展奠定基础。