LangGraph自定义AI代理2026：构建智能工作流

2026年，AI代理已经从简单的聊天机器人演变为能够执行复杂任务的智能系统。LangGraph作为构建AI代理的核心框架，提供了强大的状态管理、工具集成和工作流编排能力。本文将带你从零开始构建自定义AI代理，实现真正的任务自动化。

一、LangGraph核心概念

LangGraph基于图论思想，将AI代理建模为状态图（StateGraph）。核心概念包括：节点（Node，执行具体任务）、边（Edge，定义流转逻辑）、状态（State，存储上下文信息）。通过组合这些元素，可以构建任意复杂的工作流。

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

# Define agent state
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_action: str
    context: dict

# Initialize LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

# Define nodes
def analyze_request(state: AgentState):
    """Analyze user request and determine next action"""
    messages = state["messages"]
    response = llm.invoke([
        {"role": "system", "content": "Analyze the request and decide: research, calculate, or respond"},
        *messages
    ])
    
    action = "research" if "research" in response.content.lower() else "respond"
    
    return {
        "messages": [response],
        "next_action": action,
        "context": state["context"]
    }

def research_node(state: AgentState):
    """Perform research using tools"""
    messages = state["messages"]
    response = llm.invoke([
        {"role": "system", "content": "You are a research assistant. Use available tools to gather information."},
        *messages
    ])
    
    return {
        "messages": [response],
        "next_action": "respond",
        "context": state["context"]
    }

def respond_node(state: AgentState):
    """Generate final response"""
    messages = state["messages"]
    response = llm.invoke([
        {"role": "system", "content": "Generate a comprehensive response based on the conversation."},
        *messages
    ])
    
    return {
        "messages": [response],
        "next_action": "end",
        "context": state["context"]
    }

# Build graph
workflow = StateGraph(AgentState)

workflow.add_node("analyze", analyze_request)
workflow.add_node("research", research_node)
workflow.add_node("respond", respond_node)

workflow.set_entry_point("analyze")

workflow.add_conditional_edges(
    "analyze",
    lambda state: state["next_action"],
    {
        "research": "research",
        "respond": "respond"
    }
)

workflow.add_edge("research", "respond")
workflow.add_edge("respond", END)

# Compile
agent = workflow.compile()

# Run
result = agent.invoke({
    "messages": [{"role": "user", "content": "What are the latest AI trends in 2026?"}],
    "next_action": "",
    "context": {}
})

print(result["messages"][-1].content)

二、构建基础AI代理

基础AI代理包含三个核心节点：分析节点（理解用户意图）、执行节点（调用工具完成任务）、响应节点（生成最终答案）。通过条件边实现智能路由，让代理能够根据任务类型选择不同的处理路径。

from langgraph.graph import StateGraph
from langchain.tools import Tool
from langchain.agents import initialize_agent
from typing import TypedDict

# Define tools
def search_web(query: str) -> str:
    """Search the web for information"""
    # Implementation here
    return f"Search results for: {query}"

def calculate(expression: str) -> str:
    """Calculate mathematical expressions"""
    try:
        result = eval(expression)
        return str(result)
    except:
        return "Error in calculation"

def query_database(sql: str) -> str:
    """Query database with SQL"""
    # Implementation here
    return f"Query executed: {sql}"

# Create tools
tools = [
    Tool(
        name="WebSearch",
        func=search_web,
        description="Useful for searching the web for current information"
    ),
    Tool(
        name="Calculator",
        func=calculate,
        description="Useful for performing mathematical calculations"
    ),
    Tool(
        name="Database",
        func=query_database,
        description="Useful for querying the database with SQL"
    )
]

# Define multi-agent state
class MultiAgentState(TypedDict):
    task: str
    research_result: str
    analysis_result: str
    final_answer: str

# Research Agent
def research_agent(state: MultiAgentState):
    agent = initialize_agent(tools=[tools[0]], llm=llm)
    result = agent.run(state["task"])
    return {"research_result": result}

# Analysis Agent
def analysis_agent(state: MultiAgentState):
    agent = initialize_agent(tools=[tools[1], tools[2]], llm=llm)
    prompt = f"Analyze this research: {state['research_result']}"
    result = agent.run(prompt)
    return {"analysis_result": result}

# Synthesis Agent
def synthesis_agent(state: MultiAgentState):
    prompt = f"""
    Task: {state['task']}
    Research: {state['research_result']}
    Analysis: {state['analysis_result']}
    
    Synthesize a comprehensive answer.
    """
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# Build multi-agent workflow
workflow = StateGraph(MultiAgentState)

workflow.add_node("research", research_agent)
workflow.add_node("analysis", analysis_agent)
workflow.add_node("synthesis", synthesis_agent)

workflow.set_entry_point("research")
workflow.add_edge("research", "analysis")
workflow.add_edge("analysis", "synthesis")
workflow.add_edge("synthesis", END)

multi_agent = workflow.compile()

三、多代理协作系统

复杂任务需要多个专业代理协作。例如：研究代理负责信息收集、分析代理负责数据处理、综合代理负责生成报告。LangGraph支持构建多代理工作流，每个代理专注于特定领域，通过状态共享实现协作。

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph
import sqlite3

# Setup persistent memory
memory = SqliteSaver(sqlite3.connect("agent_memory.db"))

class PersistentAgentState(TypedDict):
    conversation_history: list
    user_preferences: dict
    learned_patterns: list

def memory_node(state: PersistentAgentState):
    """Node that manages long-term memory"""
    # Store conversation
    memory.put(
        thread_id="user_123",
        values={
            "history": state["conversation_history"],
            "preferences": state["user_preferences"]
        }
    )
    
    # Retrieve relevant memories
    relevant_memories = memory.search(
        thread_id="user_123",
        query=state["conversation_history"][-1]
    )
    
    return {
        "conversation_history": state["conversation_history"],
        "user_preferences": state["user_preferences"],
        "learned_patterns": relevant_memories
    }

def learning_node(state: PersistentAgentState):
    """Node that learns from interactions"""
    # Extract patterns from conversation
    patterns = extract_patterns(state["conversation_history"])
    
    # Update user preferences
    preferences = update_preferences(
        state["user_preferences"],
        state["conversation_history"]
    )
    
    return {
        "conversation_history": state["conversation_history"],
        "user_preferences": preferences,
        "learned_patterns": patterns
    }

# Build agent with memory
workflow = StateGraph(PersistentAgentState)
workflow.add_node("memory", memory_node)
workflow.add_node("learning", learning_node)

workflow.set_entry_point("memory")
workflow.add_edge("memory", "learning")
workflow.add_edge("learning", END)

# Compile with checkpointing
agent = workflow.compile(checkpointer=memory)

# Run with thread_id for persistence
result = agent.invoke(
    {
        "conversation_history": [{"role": "user", "content": "I prefer concise answers"}],
        "user_preferences": {},
        "learned_patterns": []
    },
    config={"configurable": {"thread_id": "user_123"}}
)

四、持久化记忆与学习

AI代理需要记住历史交互，才能提供个性化服务。LangGraph集成了持久化存储，可以保存对话历史、用户偏好和学习到的模式。这让代理能够跨会话保持上下文，越用越智能。

from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
import asyncio

class ParallelAgentState(TypedDict):
    task: str
    subtasks: list
    results: dict

async def decompose_task(state: ParallelAgentState):
    """Break down complex task into subtasks"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    response = await llm.ainvoke([
        {"role": "system", "content": "Break this task into 3-5 independent subtasks"},
        {"role": "user", "content": state["task"]}
    ])
    
    subtasks = parse_subtasks(response.content)
    return {"subtasks": subtasks, "results": {}}

async def parallel_executor(state: ParallelAgentState):
    """Execute subtasks in parallel"""
    async def execute_subtask(subtask):
        llm = ChatOpenAI(model="gpt-4-turbo")
        result = await llm.ainvoke([
            {"role": "user", "content": f"Solve this: {subtask}"}
        ])
        return subtask, result.content
    
    # Execute all subtasks concurrently
    tasks = [execute_subtask(st) for st in state["subtasks"]]
    results_list = await asyncio.gather(*tasks)
    
    results = {task: result for task, result in results_list}
    return {"results": results}

async def synthesize_results(state: ParallelAgentState):
    """Combine parallel results into final answer"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    results_text = "\n".join([
        f"Subtask: {k}\nResult: {v}" 
        for k, v in state["results"].items()
    ])
    
    response = await llm.ainvoke([
        {"role": "system", "content": "Synthesize these results into a coherent answer"},
        {"role": "user", "content": results_text}
    ])
    
    return {"final_answer": response.content}

# Build parallel workflow
workflow = StateGraph(ParallelAgentState)

workflow.add_node("decompose", decompose_task)
workflow.add_node("execute", parallel_executor)
workflow.add_node("synthesize", synthesize_results)

workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "execute")
workflow.add_edge("execute", "synthesize")
workflow.add_edge("synthesize", END)

parallel_agent = workflow.compile()

# Run
result = await parallel_agent.ainvoke({
    "task": "Analyze the impact of AI on healthcare, finance, and education in 2026",
    "subtasks": [],
    "results": {}
})

五、并行执行与性能优化

对于复杂任务，可以将子任务分解并并行执行。LangGraph支持异步执行和并发控制，大幅提升处理效率。例如，同时研究多个主题、并行查询多个数据源，最后综合结果。

from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
from typing import TypedDict

class HumanInTheLoopState(TypedDict):
    proposal: str
    human_feedback: str
    revision_count: int
    approved: bool

def generate_proposal(state: HumanInTheLoopState):
    """Generate initial proposal"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    response = llm.invoke([
        {"role": "system", "content": "Generate a detailed proposal for the given task"},
        {"role": "user", "content": state.get("task", "Default task")}
    ])
    
    return {
        "proposal": response.content,
        "revision_count": 0,
        "approved": False
    }

def human_review_node(state: HumanInTheLoopState):
    """Wait for human review - this is an interrupt point"""
    # In production, this would send notification and wait
    # For demo, we simulate human feedback
    return state

def revise_proposal(state: HumanInTheLoopState):
    """Revise proposal based on human feedback"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    response = llm.invoke([
        {"role": "system", "content": "Revise the proposal based on feedback"},
        {"role": "user", "content": f"Proposal: {state['proposal']}\nFeedback: {state['human_feedback']}"}
    ])
    
    return {
        "proposal": response.content,
        "revision_count": state["revision_count"] + 1
    }

def check_approval(state: HumanInTheLoopState):
    """Check if proposal is approved"""
    return {"approved": state.get("approved", False)}

# Build workflow with human-in-the-loop
workflow = StateGraph(HumanInTheLoopState)

workflow.add_node("generate", generate_proposal)
workflow.add_node("review", human_review_node)
workflow.add_node("revise", revise_proposal)

workflow.set_entry_point("generate")
workflow.add_edge("generate", "review")

# Conditional edge based on approval
workflow.add_conditional_edges(
    "review",
    lambda state: "end" if state.get("approved") else "revise",
    {
        "end": END,
        "revise": "revise"
    }
)

workflow.add_edge("revise", "review")

hitl_agent = workflow.compile()

# Run with human interaction
config = {"configurable": {"thread_id": "session_1"}}

# First run - generates proposal
result = hitl_agent.invoke(
    {"task": "Design a new feature for our app"},
    config=config
)

print("Proposal:", result["proposal"])

# Simulate human feedback
feedback = "Make it more user-friendly and add analytics"

# Continue with feedback
result = hitl_agent.invoke(
    {"human_feedback": feedback},
    config=config
)

print("Revised:", result["proposal"])

六、人机协作与审批流程

某些关键决策需要人工介入。LangGraph支持人机协作模式，在关键节点暂停等待人工审批，根据反馈调整方案。这确保了AI代理的决策符合业务要求，同时保持自动化效率。