07 Agent的自主规划与工具开发

AI Agent 核心概念

AI Agent 是通过大型语言模型（LLM）驱动，能够自主理解、规划和执行复杂任务的系统。给定一个目标，AI Agent 能够完成后续的全部工作。

核心组件：

规划（Planning）： 将复杂任务分解为更小、可管理的子目标。
记忆（Memory）：

Code
_短期记忆：_* 用于上下文学习，通常通过 LLM 的上下文窗口实现。 _长期记忆：_* 存储和检索历史信息，通常通过外部载体（如向量数据库）实现。
工具使用（Tool use）： 调用外部 API 或功能来获取额外信息、执行特定操作，从而增强 LLM 的能力。

Generative Agents (生成式智能体):

Google 和 Stanford 学者提出的概念，利用生成模型模拟可信的人类行为。它们能够在沙盒环境中与人、环境互动，记忆、回忆、反思观察结果，并制定日常计划。

核心模块:
1. 长期记忆： 存储 Agent 的所有经验，支持多轮决策。
2. 外部工具： 允许 Agent 调用外部服务。
3. 短期记忆： 基于注意力机制，从长期记忆中检索最相关信息喂给 LLM。

AI Agent 工具定位与对比

工具 | 核心定位 | 架构特点 | 适用场景

LangChain | 开源LLM应用开发框架 | 基于链（Chain）的线性或分支工作流，支持Agent模式 | 快速构建RAG、对话系统、工具调用等线性任务

LangGraph | LangChain的扩展，专注于复杂工作流 | 基于图（Graph）的循环和条件逻辑，支持多Agent协作 | 需要循环、动态分支或状态管理的复杂任务（如自适应RAG、多Agent系统）

Qwen-Agent | 通义千问的AI Agent框架 | 基于阿里云大模型，支持多模态交互与工具调用 | 开源，集成多种工具，MCP调用

Coze | 字节跳动的无代码AI Bot平台 | 可视化拖拽界面，内置知识库、多模态插件 | 快速部署社交平台机器人、轻量级工作流

Dify | 开源LLM应用开发平台 | API优先，支持Prompt工程与灵活编排 | 开发者定制化LLM应用，需深度集成或私有化部署

Agent 的适用场景与最佳实践

什么时候使用 Agent？

Agent 适用于那些开放性问题，这些问题所需的步骤数量难以预测，无法通过硬编码固定路径来解决。Agent 的自主性使其非常适合在受信任的环境中扩展任务。

注意事项：

高成本和错误积累： 自主性意味着更高的成本和可能出现的错误累积。建议在沙盒环境中进行广泛测试，并设置防护栏。
信任度： LLM 可能会运行多个回合，需要对其决策能力有一定信任。

打造最佳实践：

AI 智能体和工作流互补，可集成以实现最佳效果。

增强自动化： Agent 处理特定任务，工作流协调整体过程。
可扩展性： 结构化工作流结合多个 Agent 高效扩展运营。
弹性与适应性： 单个 Agent 应对局部变化，工作流动态调整整体流程。

核心原则：

保持智能体设计的简洁性： 避免不必要的复杂性，专注于核心功能。
优先考虑透明性： 明确展示智能体的规划步骤，让用户了解决策过程。
打造 Function/MCP： 打造高质量工具、说明文档和测试，确保 Agent 与外部环境的交互。
选择适合你的系统： 从简单的提示开始，只在必要时添加多步智能体系统。

反应式 (Reactive) Agent

核心概念：

反应式架构是 AI 智能体设计中最简单直接的模式，基于当前环境即时决策，无长期规划，依赖预设规则快速响应。

工作原理：

感知环境输入 -> LLM 或规则系统立即生成响应动作 -> 执行动作并观察结果 -> 循环往复直至任务完成。

优势：

速度快： 无复杂推理，适合毫秒级响应。
简单可靠： 行为由明确规则驱动，易于设计和验证。

局限：

缺乏适应性： 无法处理未预见的场景或需要多步规划的任务。
短视性： 仅优化当前动作，可能陷入局部循环。

典型应用：

机器人避障、游戏 NPC 的即时反应、工业控制中的警报触发。

案例：私募基金运作指引问答助手 (反应式)

目标： 提供高质量的私募基金规则咨询服务。

设计特点：

自主决策： 智能体根据问题类型自主选择工具。
透明思考过程： 展示清晰的决策链路。
知识边界感知： 明确区分知识库内容和模型知识。
多工具协作： 集成关键词搜索、类别查询和直接问答工具。
异常处理： 妥善处理超出知识范围的问题。

实现步骤（基于 LangChain Agent 框架与 Qwen-Turbo 模型）：

数据准备： 构建精简的私募基金运作规则知识库（FUND_RULES_DB），包含 ID、类别、问题、答案。

JSON
{ "id": "rule001", "category": "设立与募集", "question": "私募基金的合格投资者标准是什么？", "answer": "合格投资者是指具备相应风险识别能力和风险承担能力，投资于单只私募基金的金额不低于100万元且符合下列条件之一的单位和个人：\n1. 净资产不低于1000万元的单位\n2. 金融资产不低于300万元或者最近三年个人年均收入不低于50万元的个人" }
工具设计：
- search_rules_by_keywords (关键词搜索)
- search_rules_by_category (类别查询)
- answer_question (直接回答)
Agent 架构搭建：
- 自定义提示模板（CustomPromptTemplate）：定义 Agent 的思考和决策格式。
- 自定义输出解析器（CustomOutputParser）：解析 LLM 输出，确定下一步行动。
- Agent 执行器（AgentExecutor）：协调智能体与工具的交互。
知识边界处理： 识别问题主题，明确区分知识库内容和模型知识，提供有价值建议。

深思熟虑 (Deliberative) Agent

核心概念：

深思熟虑智能体基于内部模型进行规划，通过推理选择最优行动方案，具有长期目标导向性。

核心流程：

感知 (Perceive)： 获取环境信息。
建模 (Model)： 更新内部世界状态表示。
推理 (Reason)： 生成候选计划并模拟结果。
决策 (Decide)： 选择最优方案执行。

优势：

能处理多步复杂任务。
优化长期目标而非即时反馈。
适应动态变化环境。

典型示例：

路径规划智能体：生成多条候选路线，评估安全性/耗时，选择最优路径执行。

案例：智能投研助手 (深思熟虑)

目标： 整合市场数据，进行多步骤分析和推理，生成投资观点和研究报告。

设计特点：

内部建模： 构建市场模型和行业理解。
多方案生成： 基于市场模型创建多个候选投资策略。
方案评估： 对比分析各方案优劣。
长期规划： 优化整体投资回报。
推理透明： 能够解释决策过程和选择理由。

具体实现步骤（基于 LangGraph）：

感知阶段： 收集市场概况、经济指标、新闻、行业趋势。
建模阶段： 构建内部世界模型，评估市场状态、经济周期、风险机会。
推理阶段： 生成多个候选投资分析方案，包含假设、方法、预期结果等。
决策阶段： 评估方案并选择最优投资观点，形成投资论点、证据、风险评估、建议。
报告阶段： 生成完整的投资研究报告。

智能体状态定义 (ResearchAgentState)：

Python

from typing import TypedDict, Optional, Dict, Any, List, Literal

class ResearchAgentState(TypedDict):
    # 输入
    research_topic: str  # 研究主题
    industry_focus: str  # 行业焦点
    time_horizon: str  # 时间范围(短期/中期/长期)
    # 处理状态
    perception_data: Optional[Dict[str, Any]]  # 感知阶段收集的数据
    world_model: Optional[Dict[str, Any]]  # 内部世界模型
    reasoning_plans: Optional[List[Dict[str, Any]]]  # 候选分析方案
    selected_plan: Optional[Dict[str, Any]]  # 选中的最优方案
    # 输出
    final_report: Optional[str]  # 最终研究报告
    # 控制流
    current_phase: Literal["perception", "modeling", "reasoning", "decision", "report"]
    error: Optional[str]  # 错误信息

工作流实现 (create_research_agent_workflow)：

Python

from langgraph.graph import StateGraph, END
# ... other imports for nodes like perception, modeling etc.

def create_research_agent_workflow() -> StateGraph:
    workflow = StateGraph(ResearchAgentState)

    workflow.add_node("perception", perception)
    workflow.add_node("modeling", modeling)
    workflow.add_node("reasoning", reasoning)
    workflow.add_node("decision", decision)
    workflow.add_node("report", report_generation)

    workflow.set_entry_point("perception")
    workflow.add_edge("perception", "modeling")
    workflow.add_edge("modeling", "reasoning")
    workflow.add_edge("reasoning", "decision")
    workflow.add_edge("decision", "report")
    workflow.add_edge("report", END)

    return workflow.compile()

各阶段处理逻辑示例 (推理阶段 reasoning)：

Python

import json
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
# Assume 'llm' and 'REASONING_PROMPT' are defined elsewhere

def reasoning(state: ResearchAgentState) -> ResearchAgentState:
    try:
        if not state.get("world_model"):
            return {
                **state,
                "error": "推理阶段缺少世界模型",
                "current_phase": "modeling"
            }

        prompt = ChatPromptTemplate.from_template(REASONING_PROMPT)
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "world_model": json.dumps(state["world_model"], ensure_ascii=False, indent=2)
        }        
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)

        return {
            **state,
            "reasoning_plans": result,
            "current_phase": "decision"
        }
    except Exception as e:
        return {
            **state,
            "error": f"推理阶段出错: {str(e)}",
            "current_phase": "reasoning"
        }

混合式 (Hybrid) Agent

核心概念：

混合智能体架构结合了反应式的“快速本能”和深思熟虑的“战略规划”，旨在实现智能与效率的平衡。

三层设计：

底层（反应式）： 即时处理紧急任务（如避障），快速响应。
中层（协调）： 管理任务优先级，动态选择处理模式（反应式或深思熟虑）。
顶层（深思熟虑）： 进行长期目标规划（如路径优化），深度思考。

运作机制：

通过仲裁系统（如监督器）动态切换模式：

紧急情况 → 启用反应式快速响应。
常规情况 → 启动深思熟虑规划。

核心优势：

兼具实时响应能力和战略规划优势。

典型示例：

自动驾驶车辆：突发障碍时立即刹车（反应式），正常行驶时规划最优路线（深思熟虑）。

案例：投顾AI助手 (混合式)

目标： 提供智能化、个性化的财富管理咨询服务。

架构设计：

底层（反应式层）： 处理简单直接的查询（市场状况、账户信息），提供毫秒级反馈。
中层（协调层）： 评估客户查询类型和优先级，动态选择反应式或深思熟虑处理模式。
顶层（深思熟虑层）： 处理复杂的投资分析和长期财务规划，构建内部模型并生成多个备选方案。

处理流程：

查询评估阶段（协调层）： 评估查询类型（紧急、信息、分析）和处理模式（反应式或深思熟虑）。
反应式处理流程： 针对简单查询，低延迟、高响应速度，直接调用数据和预设回答。
深思熟虑处理流程： 针对复杂分析型查询，进行数据收集、深度分析、生成建议。

智能体状态管理 (WealthAdvisorState)：

Python

from typing import TypedDict, Optional, Dict, Any, Literal

class WealthAdvisorState(TypedDict):
    # 输入
    user_query: str  # 用户查询
    customer_profile: Optional[Dict[str, Any]]  # 客户画像
    # 处理状态
    query_type: Optional[Literal["emergency", "informational", "analytical"]]  # 查询类型
    processing_mode: Optional[Literal["reactive", "deliberative"]]  # 处理模式
    emergency_response: Optional[Dict[str, Any]]  # 紧急响应结果
    market_data: Optional[Dict[str, Any]]  # 市场数据
    analysis_results: Optional[Dict[str, Any]]  # 分析结果
    # 输出
    final_response: Optional[str]  # 最终响应
    # 控制流
    current_phase: Literal["assess", "reactive", "collect_data", "analyze", "recommend", "respond"]
    error: Optional[str]  # 错误信息

使用场景示例：

反应式： 查询“今天上证指数的表现如何？” → 协调层判断为“紧急/反应式” → 反应式层快速查询并返回市场数据。
深思熟虑： 查询“我应该如何调整投资组合以应对可能的经济衰退？” → 协调层判断为“分析型/深思熟虑” → 深思熟虑层进行多步分析、生成详细投资建议报告。

LangGraph 使用详解

LangGraph 是 LangChain 的扩展，专注于复杂工作流的构建，通过图形化方式支持循环和条件逻辑。

ChatTongyi 的作用：

ChatTongyi 是 LangChain 社区提供的通义千问模型封装类，用于与 LLM 交互。它支持 bind_tools() 方法，允许 LLM 理解并调用工具。

Python
from langchain_community.chat_models import ChatTongyi llm = ChatTongyi(model_name="qwen-turbo-latest", dashscope_api_key="YOUR_API_KEY") # 绑定工具到LLM（关键步骤） llm_with_tools = llm.bind_tools(tools)
LangGraph 中的工具调用：

使用 @tool 装饰器定义工具函数。LLM 分析用户查询后决定是否调用工具，ToolNode 执行工具，返回结果，LLM 基于结果生成响应。

Python
from langchain_core.tools import tool @tool def query_shanghai_index() -> str: """查询上证指数实时行情，获取当前点位、涨跌和涨跌幅信息""" name = "上证指数" price = "3125.62" change = "6.32" pct = "0.20" result = f"{name} 当前点位: {price}，涨跌: {change}，涨跌幅: {pct}%" return result tools = [query_shanghai_index] # 可以添加更多工具

工具调用是一个循环过程，LLM 可能多次调用不同工具直到获取足够信息。
ToolNode 的作用：

ToolNode 是 LangGraph 预构建的节点，专门用于执行工具调用。它接收包含工具调用请求的消息，执行工具函数，并将结果封装为 ToolMessage 更新到状态中。

Python
from langgraph.prebuilt import ToolNode tool_node = ToolNode(tools) # 将ToolNode添加到工作流中 # workflow.add_node("tools", tool_node)
StateGraph 的作用：

StateGraph 是 LangGraph 的核心类，用于定义和管理 Agent 的状态转换图。它允许构建复杂的、有状态的工作流，其中每个节点可以读取和更新共享状态。状态通过 TypedDict 定义，提供类型安全。

功能 | 说明 | 代码示例

状态管理 | 维护整个工作流的共享状态，节点可读取更新 | WealthAdvisorState 定义

节点定义 | 将函数封装为工作流节点 | add_node()

条件路由 | 根据状态动态决定下一个节点 | add_conditional_edges()

固定路由 | 节点之间的固定连接 | add_edge()

入口点 | 定义工作流的起始节点 | set_entry_point()

LangGraph 使用步骤总结：

定义状态： 使用 TypedDict 定义工作流的状态结构。
创建 StateGraph： 实例化 StateGraph 并传入状态类型。
定义节点函数： 编写处理逻辑函数，接收状态并返回状态更新。
添加节点： 使用 add_node() 将函数添加到工作流。
设置入口点： 使用 set_entry_point() 指定起始节点。
添加边： 使用 add_edge() 或 add_conditional_edges() 连接节点。
编译工作流： 调用 compile() 方法编译工作流。
执行工作流： 调用 invoke() 方法传入初始状态执行。

构建 Agent 的核心思想总结

不要为所有任务构建 Agent：

Code
_适用场景：_* Agent 适合处理复杂、模糊且高价值的任务，而非所有场景。 _判断标准：_* 任务复杂性、任务价值、关键能力验证、错误成本。 _案例：_* 代码生成因其复杂性高、价值大且输出易验证（如单元测试）是理想场景。
保持简洁：

Code
_Agent 的核心组件：_* 环境、工具集、系统提示。 _设计原则：_* 初期避免过度复杂化，优先迭代核心组件。优化可在基础行为稳定后进行。 _案例：_* 不同功能的 Agent 可共享相同代码框架，仅调整工具和提示。
像 Agent 一样思考：

Code
_理解 Agent 的局限性：_* Agent 仅基于有限上下文做决策。 _改进方法：_* 模拟 Agent 的视角（如仅通过静态截图操作电脑）以发现设计缺陷。直接询问模型以验证指令清晰度或工具使用合理性。分析轨迹日志，优化上下文提供方式。

核心目标： 在提升 Agent 能力的同时，平衡成本、风险与用户体验。

关键要点

AI Agent 通过规划、记忆和工具使用实现复杂任务的自主执行。
根据任务特性，选择反应式（即时响应）、深思熟虑（长期规划）或混合式（结合两者）架构。
LangGraph 是构建复杂 Agent 工作流的强大框架，支持状态管理和图式编排。
构建 Agent 时应遵循简洁性、透明性和功能/工具完整性原则。
并非所有任务都适合 Agent，需评估任务的复杂性、价值和潜在风险。

embed