企业级Agent智能体(智能小秘)之LangGraph智能体

🧠 智能体模式的核心差异

🔄 用ReAct模式实现的真正智能体

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
import json# 1. 定义状态 - 增加更多的思维痕迹
class AgentState(TypedDict):user_query: str           # 用户原始问题thoughts: List[str]      # 思维链：记录推理过程next_action: str         # 下一步要做什么available_tools: List[str] # 可用工具列表tool_input: dict        # 工具输入参数tool_output: str        # 工具执行结果final_answer: str       # 最终答案needs_human: bool       # 是否需要人工介入
}# 2. 定义 定义工具集 - 智能体的"技能"
def search_knowledge_base(query: str) -> str:"""搜索知识库工具"""print(f"🔍 正在知识库中搜索: {query}")# 模拟搜索结果knowledge = {"billing": "计费问题通常涉及扣款金额、周期、退款等。请联系财务部门核实具体情况。""technical": "技术问题包括登录失败、页面错误、功能异常等。请尝试清除缓存或更换浏览器。"，"account": "账户问题包含注册、登录、注销、信息安全等。请确认操作步骤是否正确。"}return knowledge.get(query.split()[0], "未找到相关信息")def check_order_status(order_ref: str) -> str:"""检查订单状态工具"""print(f"📦 正在查询订单: {order_ref}")return f"订单 {order_ref} 状态: 已发货，预计明天送达"def escalate_to_human(reason: str) -> str:"""升级到人工工具"""print(f"🚨 正在升级到人工，原因:原因: {reason}")return "问题已分配给人工客服，请稍候..."# 3. 核心核心智能体节点 - 具备思考和决策能力
def reasoning_agent(state: AgentState):"""推理智能体：分析和决定下一步行动"""from openai import OpenAIclient = OpenAI()# 构建可用工具的描述tools_description = """
可用工具:
1. search_knowledge_base(query): 搜索知识库获取解决方案
2. check_order_status(order_ref): 查询特定订单状态
3. escalate_to_human(reason): 升级到人工客服
"""# 构建思维提示词system_prompt = """
你是一个技术支持智能体。请按照以下步骤思考:1. 分析用户问题的本质
2. 评估现有知识和工具是否能解决问题  
3. 如果能，选择合适的工具并提供必要的输入参数
4. 如果不能，决定升级到人工客服请用JSON格式回应:
{"thought": "你的推理过程","next_action": "think|tool_name|respond|escalate","tool_name": "要使用的工具名(如果是tool)","tool_input": {"param": "value"}, # 工具输入参数"response": "如果要直接回答的内容"
}
"""# 获取之前的上下文conversation_history = "\n".join([f"- {t}" for t in state.get("thoughts", [])])response = client.chat.completions.create(model="gpt-4",messages=[{"role": "system", "content": system_prompt},{"role": "user", "content": f"用户问题: {state['user_query']}\n\n之前的思考:{conversconversation_history}\n\n{tools_description}"}],response_format={"type": "json_object"})decision = json.loads(response.choices[0].message.content)# 更新状态 - 记录 记录思维轨迹current_thoughts = state.get("thoughts", [])current_thoughts.append(decision["thought"])state["thoughts"] = current_thoughtsstate["next_action"] = decision["next_action"]if decision["next_action"] == "tool":state["tool_name"] = decision["tool_name"]state["tool_input"] = decision["tool_input"]return state# 4. 工具执行节点
def tool_executor(state: AgentState):"""工具执行器：根据智能体的决策调用具体工具"""tool_name = state["tool_name"]tool_input = state["tool_input"]print(f"🛠️ 正在执行工具: {tool_name}, 输入: {tool_input}")# 根据工具名分发执行if tool_name == "search_knowledge_base":result = search_knowledge_base(tool_input["query"])elif tool_name == "check_order_status":result = check_order_status(tool_input["order_ref"])  elif tool_name == "escalate_to_human":result = escalate_to_human(tool_input["reason"])state["tool_output"] = resultreturn state# 5. 响应生成节点
def response_generator(state: AgentState):"""响应生成器：综合所有信息生成最终回答"""# 如果有工具输出，结合工具结果生成回答if "tool_output" in state and state["tool_output"]:state["final_answer"] = f"根据我们的分析: {state['tool_output']}"else:# 直接从推理结果中提取state["final_answer"] = state.get("response", "抱歉，我暂时无法处理这个问题。")return state# 6. 构建智能体工作流
def create_agentic_support_system():builder = StateGraph(AgentState)# 添加节点builder.add_node("reasoning_agent", reasoning_agent)builder.add_node("tool_executor", tool_executor)builder.add_node("response_generator", response_generator)# 设置入口点builder.set_entry_point("reasoning_agent")# 添加条件边 - 这才是真正的"智能"所在!def should_continue(state: AgentState):"""决定下一步行动的智能路由"""next_action = state["next_action"]if next_action == "tool":return "use_tool"elif next_action == "respond":return "generate_response"elif next_action == "escalate":state["needs_human"] = Truereturn "generate_response"else:return "continue_reasoning"builder.add_conditional_edges("reasoning_agent",should_continue,{"use_tool": "tool_executor","generate_response": "response_generator","continue_reasoning": "reasoning_agent"})# 从工具执行器返回到推理智能体builder.add_edge("tool_executor", "reasoning_agent")# 从响应生成器结束builder.add_edge("response_generator", END)return builder.compile()# 7. 使用示例
if __name__ == "__main__":# 创建智能体系统agent_system = create_agentic_support_system()# 测试用例test_cases = ["我的信用卡扣款金额不对","订单号ORD123456现在什么状态？","我忘记密码了怎么办","紧急！系统崩溃了，我要投诉！"]for i, query in enumerate(test_cases, 1):print(f"\n{'='*60}")print(f"🧪 测试用例 {i}: {query}")print(f"{'='*60}")# 初始化状态initial_state = {"user_query": query,"thoughts": [],"available_tools": ["search_knowledge_base", "check_order_status", "escalate_to_human"]}# 执行智能体系统try:result = agent_system.invoke(initial_state)print("\n🤖 智能体执行轨迹:")for j, thought in enumerate(result["thoughts"], 1):print(f"  {j}. {thought}")print(f"\n💬 最终回答: {result['final_answer']}")print(f"👤 需要人工: {result.get('needs_human', False)}")except Exception as e:print(f"❌ 执行出错: {e}")

🏗️ 智能体架构的核心思想

传统的if-else工作流：

 用户输入 → if A then X → if B then Y → end

智能体工作流：

 用户输入 → 思考该做什么 → 执行 → 观察结果 → 再思考 → ... → 满意后结束

🔍 第一步：状态设计 - 智能体的"记忆"

class AgentState(TypedDict):user_query: str           # 用户原始问题thoughts: List[str]      # 思维链：记录推理过程next_action: str         # 下一步要做什么available_tools: List[str] # 可用工具列表tool_input: dict        # 工具输入参数tool_output: str        # 工具执行结果final_answer: str       # 最终答案needs_human: bool       # 是否需要人工介入
}

🤔 为什么这样设计状态？

• thoughts: List[str]：

  • 这是思维链的具体体现
  • 记录智能体每一步的推理过程，便于调试和理解
  • 举例：[“用户问扣款问题，这是计费类”, “我可以搜索知识库找解决方案”]

• next_action: str：

  • 这是智能体的"决策输出"
  • 可选值：think|tool|respond|escalate
  • 定义了智能体所有可能的行为类型

• tool_input: dict：

  • 通用的参数容器，可以适应任何工具
  • 比如：{“query”: “信用卡扣款”}, {“order_ref”: “123”}

• available_tools：

  • 让智能体知道它有哪些"技能"可用

🛠️ 第二步：工具定义 - 智能体的"技能"

def search_knowledge_base(query: str) -> str:"""搜索知识库工具"""print(f"🔍 正在知识库中搜索: {query}")# 模拟搜索逻辑knowledge = {"billing": "计费问题通常涉及扣款金额、周期、退款等..."}return knowledge.get(query.split()[0], "未找到相关信息")

🤔 为什么工具要这样设计？

• 函数即工具：每个工具都是一个Python函数
• 明确的接口：输入参数和返回值都很清晰
• 现实世界连接：这些工具可以替换成真实的API调用

🧠 第三步：核心推理智能体 - 智能体的"大脑"

这是最关键的部件，让我们仔细分析：

def reasoning_agent(state: AgentState):# 1. 准备工具描述 - 告诉智能体它有什么能力tools_description = """
可用工具:
1. search_knowledge_base(query): 搜索知识库获取解决方案
2. check_order_status(order_ref): 查询特定订单状态  
3. escalate_to_human(reason): 升级到人工客服
"""

🎯 系统提示词设计

system_prompt = """
你是一个技术支持是一个技术支持智能体。请按照以下步骤思考:1. 分析用户问题的本质
2. 评估现有知识和工具是否能解决问题是否能解决问题  
3. 如果能，选择合适的工具并提供必要的输入参数
4. 如果不能，决定升级到人工客服请用JSON格式回应:
{"thought": "你的推理过程","next_action": "think|tool_name|respond|escalate","tool_name": "要使用的工具名(如果是tool)","tool_input": {"param": "value"}, # 工具输入参数"response": "如果要直接回答的内容"
}
"""

🤔 为什么提示词要这样设计？

• 结构化思考：强制智能体按步骤分析
• JSON格式化：确保输出的机器可读性
• 明确的字段：每个字段都有特定用途

🔄 第四步：构建工作流图 - 智能体的"行为模式"

这是LangGraph的精髓所在：

4.1 决策函数 - 智能体的"直觉"

def should_continue(state: AgentState):"""决定下一步行动的智能路由"""next_action = state["next_action"]if next_action == "tool":return "use_tool"          # 需要执行工具elif next_action == "respond":return "generate_response" # 可以直接回答elif next_action == "escalate":state["needs_human"] = Truereturn "generate_response"else:return "continue_reasoning" # 还需要继续思考

🎯 这个函数的运作逻辑：

 接收当前状态 → 查看next_action字段 → 根据值决定下一步去哪 → 返回目标节点名称

4.2 图形构建过程

# 创建图形框架
builder = StateGraph(AgentState)# 添加节点 - 相当于组建团队
builder.add_node("reasoning_agent", reasoning_agent)  # 军师
builder.add_node("tool_executor", tool_executor)    # 工匠
builder.add_node("response_generator", response_generator) # 发言人# 设置谁第一个发言
builder.set_entry_point("reasoning_agent")# 建立沟通渠道
builder.add_conditional_edges("reasoning_agent",       # 从谁那里发出消息should_continue,         # 用什么逻辑决定传给谁{                        # 可能的接收者名单"use_tool": "tool_executor","generate_response": "response_generator","continue_reasoning": "reasoning_agent"}
)# 建立固定链接 - 工匠做完活一定要向军师汇报
builder.add_edge("tool_executor", "reasoning_agent")# 发言人说完就结束
builder.add_edge("response_generator", END)

🚀 第五步：执行流程详解

让我们追踪一个具体例子的完整执行过程：

输入：

initial_state = {"user_query": "我的信用卡扣款金额不对","thoughts": [],  # 刚开始，没有任何思考记录"available_tools": ["search_knowledge_base", "check_order_status", "escalate_to_human"]
}

第1轮执行：

1.推理智能体工作：

• 输入：只有用户问题，thoughts为空
• 过程：分析问题 → 发现是计费问题 → 决定搜索知识库

输出JSON：

{"thought": "用户在询问信用卡扣款问题，这明显是个计费类问题，我应该先用search_knowledge_base工具查找相关解决方案""next_action": "tool","tool_name": "search_knowledge_base","tool_input": {"query": "信用卡扣款金额不对"}
}

此时状态变为：

{"user_query": "我的信用卡扣款金额不对","thoughts": ["用户在询问信用卡扣款问题，这明显是个计费类问题..."],"next_action": "tool","tool_name": "search_knowledge_base","tool_input": {"query": "信用卡扣款金额不对"},"final_answer": "","needs_human": False
}

决策函数判断：

• next_action = “tool”
• 返回：“use_tool”

第2轮执行：

2.工具执行器工作：

输入：知道要调用search_knowledge_base，参数是{“query”: “信用卡扣款金额不对”}

执行结果：

state["tool_output"] = "计费问题通常涉及扣款金额、周期、退款等。请联系财务部门核实具体情况。"#状态变化：
{... # 保持其他字段不变"tool_output": "计费问题通常涉及扣款金额、周期、退款等。请联系财务部门核实具体情况。"
}

由于建立了固定边：add_edge(“tool_executor”, “reasoning_agent”)
→ 自动返回到推理智能体

第3轮执行：

3.推理智能体再次工作：

• 输入：现在有了工具执行的结果
• 新的思考：“搜索结果显示这是标准的计费流程问题，需要联系财务部门。我可以直接给用户这个建议。”

输出JSON：

{"thought": "知识库给出了明确的指引，我可以直接回答用户了","next_action": "respond","response": "根据我们的知识库，您的信用卡扣款问题需要联系财务部门核实具体情况。他们会帮助您解决。"
}

决策函数再次判断：

• next_action = “respond”
• 返回：“generate_response”

第4轮执行：

4.响应生成器工作：

• 综合所有信息生成友好回答
• 状态：`“final_answer”: “根据我们的分析，您的信用卡扣款问题需要联系财务部门核实具体情况…”

---
## 🎯 智能体决策的四种路径### 路径1：思考 → 工具 → 思考 → 响应 ✅