LangGraph教程：用Python构建有状态AI代理

from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# 1. 定义代理状态
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    current_task: str
    tool_output: str

# 2. 初始化LLM
llm = ChatOpenAI(model="gpt-5-chat", temperature=0)

# 3. 定义LLM节点
def call_model(state: AgentState):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# 4. 定义工具节点 (示例，实际工具需实现)
def call_tool(state: AgentState):
    # 假设这里调用一个搜索工具
    print(f"执行工具进行搜索: {state['current_task']}")
    tool_result = "搜索结果：LangGraph是一个用于构建有状态AI代理的框架。"
    return {"tool_output": tool_result, "messages": [HumanMessage(content=tool_result)]}

# 5. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("llm", call_model)
workflow.add_node("tool", call_tool)

workflow.add_edge(START, "llm")

# 条件路由：如果LLM响应需要工具，则转到工具节点
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if "需要工具" in last_message.content:
        return "tool"
    else:
        return "end"

workflow.add_conditional_edges("llm", should_continue, {"tool": "tool", "end": END})
workflow.add_edge("tool", END) # 工具执行完直接结束，或返回LLM继续处理

# 编译图
app = workflow.compile()

# 运行代理
inputs = {"messages": [HumanMessage(content="请告诉我LangGraph是什么，并模拟一次工具搜索。")],
          "current_task": "查询LangGraph信息"}

for s in app.stream(inputs):
    print(s)
    print("----")

# 更复杂的运行，模拟LLM需要工具
inputs_with_tool_request = {"messages": [HumanMessage(content="请使用工具查询最新的AI发展。我需要工具。")],
                            "current_task": "查询最新AI发展"}

print("\n--- 模拟LLM需要工具的场景 ---")
for s in app.stream(inputs_with_tool_request):
    print(s)
    print("----")

from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

# 1. 定义更复杂的代理状态，包含任务、代码和测试结果
class CodeAgentState(TypedDict):
    task: str
    code: Annotated[str, operator.setitem]
    test_results: Annotated[List[str], operator.add]
    iterations: int

# 2. 初始化LLM (假设GPT-5.3-Codex用于代码生成)
code_llm = ChatOpenAI(model="gpt-5-3-codex", temperature=0.2)

# 3. 定义代码生成节点
def generate_code(state: CodeAgentState):
    prompt = f"根据任务 '{state['task']}' 生成Python代码。当前代码：{state.get('code', '无')}"
    response = code_llm.invoke([HumanMessage(content=prompt)])
    print(f"生成代码: {response.content[:50]}...")
    return {"code": response.content, "iterations": state['iterations'] + 1}

# 4. 定义代码测试节点 (模拟执行和测试)
def test_code(state: CodeAgentState):
    code = state["code"]
    # 模拟执行代码并返回测试结果
    if "def add(a, b):" in code and state['iterations'] < 3: # 模拟前几次测试失败
        result = "测试失败：加法函数未正确处理负数。请修改。"
        print(f"测试结果：{result}")
        return {"test_results": [result]}
    else:
        result = "测试通过：所有测试用例均通过。"
        print(f"测试结果：{result}")
        return {"test_results": [result]}

# 5. 构建图
workflow = StateGraph(CodeAgentState)

workflow.add_node("generate_code", generate_code)
workflow.add_node("test_code", test_code)

workflow.add_edge(START, "generate_code")
workflow.add_edge("generate_code", "test_code")

# 条件路由：根据测试结果决定是否循环
def should_loop(state: CodeAgentState):
    if "测试失败" in state["test_results"][-1] and state["iterations"] < 5:
        print("测试失败，返回重新生成代码...")
        return "generate_code"
    else:
        print("测试通过或达到最大迭代次数，结束。")
        return END

workflow.add_conditional_edges("test_code", should_loop, {"generate_code": "generate_code", END: END})

# 编译图
app = workflow.compile()

# 运行代理
initial_state = {"task": "编写一个Python函数，实现两个数字的加法，并能处理正负数", 
                 "code": "", 
                 "test_results": [], 
                 "iterations": 0}

print("\n--- 运行代码生成与测试代理 ---")
for s in app.stream(initial_state):
    print(s)
    print("----")