AI提示词工程实战：Prompt格式设计与多轮对话状态管理

AI智能体（Agent）在实际业务场景中往往需要完成多步骤、有依赖关系的复杂任务，单纯依靠大模型的单轮对话能力难以满足需求。LangGraph作为LangChain推出的工作流编排框架，将智能体行为抽象为有向图结构，支持循环、条件分支、人工介入等复杂控制流，适合构建具备自主规划与工具调用能力的AI智能体系统。

LangGraph核心概念与图结构设计

LangGraph的底层模型是StateGraph，每个节点是一个函数，接收当前状态并返回状态更新。状态通过TypedDict定义，在节点间传递。边分为普通边和条件边：普通边定义固定的节点跳转路径，条件边根据状态内容动态决定下一个执行节点。

以下是一个多步骤任务规划智能体的状态定义与图结构示例：

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

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    current_step: str
    task_complete: bool
    tool_results: List[str]

def planning_node(state: AgentState) -> dict:
    # 任务规划节点：分析用户请求，生成执行步骤
    messages = state["messages"]
    plan = llm.invoke([
        SystemMessage(content="task planner: decompose user request into steps"),
        *messages
    ])
    return {"messages": [plan], "current_step": "execute"}

def execute_node(state: AgentState) -> dict:
    # 工具执行节点：调用外部工具完成具体任务
    last_message = state["messages"][-1]
    tool_calls = parse_tool_calls(last_message)
    results = []
    for call in tool_calls:
        result = execute_tool(call)
        results.append(result)
    return {"tool_results": results, "current_step": "review"}

def review_node(state: AgentState) -> dict:
    # 结果审查节点：判断任务是否完成
    if check_completion(state["tool_results"]):
        return {"task_complete": True, "current_step": END}
    return {"current_step": "planning"}

def should_continue(state: AgentState) -> str:
    # 条件边路由函数
    if state["task_complete"]:
        return END
    return state["current_step"]

# 构建图
graph = StateGraph(AgentState)
graph.add_node("planning", planning_node)
graph.add_node("execute", execute_node)
graph.add_node("review", review_node)

graph.set_entry_point("planning")
graph.add_edge("planning", "execute")
graph.add_edge("execute", "review")
graph.add_conditional_edges("review", should_continue, {
    "planning": "planning",
    "execute": "execute",
    END: END
})

app = graph.compile()

工具调用与函数绑定机制

LangGraph本身不直接管理工具调用，工具绑定依赖LangChain的tool calling机制。大模型通过function calling能力识别需要调用的工具，LangGraph节点负责执行工具并将结果回传到状态中。

工具定义使用@tool装饰器，框架自动从函数签名和docstring生成工具描述：

from langchain_core.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search internal knowledge base and return matching documents."""
    results = vector_store.similarity_search(query, k=limit)
    return format_results(results)

@tool
def execute_sql(query: str) -> str:
    """Execute SQL query and return results."""
    try:
        result = db_engine.execute(query)
        return result.to_markdown()
    except Exception as e:
        return f"Query failed: {e}"

@tool
def send_notification(channel: str, message: str) -> str:
    """Send notification to specified channel."""
    notifier.send(channel, message)
    return "Notification sent"

# 绑定工具到LLM
llm_with_tools = llm.bind_tools([
    search_database,
    execute_sql,
    send_notification
])

人工介入与中断恢复机制

在涉及敏感操作（如删除数据、发送通知）时，智能体需要在执行前获取人工确认。LangGraph通过interrupt机制实现中断点，图执行暂停并等待外部输入，恢复后继续执行。

from langgraph.checkpoint.memory import MemorySaver

# 配置检查点存储
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# 配置中断点：在执行节点前暂停
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["execute"]
)

# 首次执行，在execute节点前暂停
config = {"configurable": {"thread_id": "task-001"}}
result = app.invoke(initial_state, config)

# 人工审核后恢复执行
result = app.invoke(None, config)

每个thread_id对应一个独立的执行会话，检查点存储记录状态的完整快照。生产环境建议使用PostgreSQL或Redis作为持久化后端，避免内存存储在服务重启后丢失状态。

多智能体协作与消息传递

复杂任务可能需要多个专业智能体协作完成。LangGraph支持创建多智能体图，不同智能体负责不同领域，通过共享状态或消息队列进行通信。

def research_agent(state: AgentState) -> dict:
    # 研究智能体：负责信息收集与分析
    findings = conduct_research(state["messages"])
    return {"messages": [HumanMessage(content=findings)]}

def writing_agent(state: AgentState) -> dict:
    # 写作智能体：基于研究结果生成报告
    draft = generate_report(state["messages"])
    return {"messages": [AIMessage(content=draft)]}

def review_agent(state: AgentState) -> dict:
    # 审核智能体：检查报告质量与准确性
    feedback = review_report(state["messages"][-1].content)
    if feedback["pass"]:
        return {"task_complete": True, "current_step": END}
    return {"messages": [HumanMessage(content=feedback["comments"])]}

# 构建多智能体协作图
multi_graph = StateGraph(AgentState)
multi_graph.add_node("research", research_agent)
multi_graph.add_node("writing", writing_agent)
multi_graph.add_node("review", review_agent)
multi_graph.set_entry_point("research")
multi_graph.add_edge("research", "writing")
multi_graph.add_edge("writing", "review")
multi_graph.add_conditional_edges("review", should_continue, {
    "research": "research",
    END: END
})

多智能体架构的关键设计点在于状态共享机制。全量共享状态会导致上下文膨胀，影响模型推理质量。实践中可采用增量传递策略，只传递必要的中间结果而非完整消息历史。

生产部署与性能考量

LangGraph应用部署到生产环境需要关注几个关键问题。图的编译结果是一个Runnable对象，可以通过LangServe部署为API服务。并发场景下每个请求使用独立的thread_id，确保状态隔离。

from langserve import add_routes
from fastapi import FastAPI

app_fastapi = FastAPI()
add_routes(app_fastapi, app, path="/agent")

# 持久化检查点配置（PostgreSQL）
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)
app = graph.compile(checkpointer=checkpointer)

对于长耗时任务，同步调用会阻塞HTTP连接。推荐使用异步模式配合任务队列，客户端通过轮询或WebSocket获取执行结果。状态图的执行本身是同步的，可将整个执行过程包装在异步任务中：

from celery import Celery

celery_app = Celery("agent_tasks", broker="redis://localhost:6379")

@celery_app.task
def run_agent_task(thread_id: str, user_input: str):
    state = create_initial_state(user_input)
    config = {"configurable": {"thread_id": thread_id}}
    result = app.invoke(state, config)
    return result

监控方面，LangGraph的每次节点执行可以通过回调机制记录耗时和状态变化，接入OpenTelemetry或LangSmith进行链路追踪，定位执行瓶颈和异常节点。