Agent智能体系统的设计与应用：从理论到实践

本文围绕Agent智能体系统展开，系统覆盖定义、核心特性、主流框架对比、三类经典架构（反应式、深思熟虑式、混合式）及对应实操案例，最后提炼构建Agent的核心思想，形成“理论-架构-案例-方法论”的完整知识体系，以下从七大模块详细整理内容：

深思熟虑智能体：基于 tencent youtu-agent 的五阶段投资研究系统https://blog.csdn.net/chenchihwen/article/details/151016532?ops_request_misc=%257B%2522request%255Fid%2522%253A%252218e0ad6f5920d560b3362c9b68a99f87%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=18e0ad6f5920d560b3362c9b68a99f87&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-3-151016532-null-null.nonecase&utm_term=%E6%8A%95%E8%B5%84&spm=1018.2226.3001.4450

一、Agent智能体的核心定义与本质

1. 基本定义

AI Agent是通过大语言模型（LLM）实现自主理解、规划、执行复杂任务的系统：只需给定目标，Agent即可自主拆解任务、调用工具、管理记忆，无需人工干预完成全流程。

2. 三大核心能力：实现“自主化”的关键

Agent区别于传统自动化工具的核心，在于具备类人类的决策与执行能力，具体通过三大模块支撑：

核心能力	定义与作用	实现方式
规划（Planning）	将复杂目标拆解为可管理的子任务，避免“一步到位”的决策压力，确保任务有序推进	基于LLM的“思维链（Chain of Thoughts）”，生成子任务优先级、执行顺序（如“先查天气→再定活动时间”）
记忆（Memory）	存储与检索任务相关信息，支撑多轮决策与长期目标优化	- 短期记忆：依赖LLM上下文窗口，处理当前对话/任务的即时信息； - 长期记忆：通过外部载体（如数据库、向量库）存储，需时快速检索（如用户历史投资偏好）
工具使用（Tool use）	调用外部API、服务或工具，弥补LLM“静态知识”与“执行能力”的短板	标准化工具接口（如MCP协议、Function Calling），LLM自主判断何时调用何种工具（如查天气用高德API、搜新闻用Bing接口）

3. 从“自动化”到“自主化”：Generative Agents的突破

2023年斯坦福与Google团队提出的Generative Agents（生成式智能体），标志着Agent从“机械执行”迈向“模拟人类行为”，核心成果体现在：

• 模拟可信人类行为：在“Smallville小镇”沙盒中，25个Agent可自主完成“日常作息（起床、工作）、社交互动（聊天、分享新闻）、长期规划（制定日程）”，甚至形成群体行为（如某Agent竞选镇长，消息在小镇传播）。

• 核心支撑：记忆流（Memory Stream）：全面记录Agent的所有经验（如与他人对话、环境互动），以自然语言存储，供LLM推理时检索，确保行为的连贯性与逻辑性（如Agent后续对话能回忆起此前相遇时的话题）。

二、主流AI Agent工具对比：定位、架构与适用场景

不同Agent工具的设计目标与架构差异显著，需根据任务复杂度与场景需求选择，核心对比如下：

工具名称	核心定位	架构特点	适用场景
LangChain	开源LLM应用开发框架（基础工具集）	基于“链（Chain）”的线性/分支工作流，支持单Agent的工具调用与RAG（检索增强生成）	快速构建简单任务（如单轮对话机器人、文档问答、天气查询），适合新手入门
LangGraph	LangChain扩展，专注复杂工作流与多Agent协作	基于“图（Graph）”的循环与条件逻辑，支持状态管理与多Agent动态交互	复杂任务（如自适应RAG、多Agent分工分析、流程化投研报告生成）
Qwen-Agent	阿里通义千问开源AI Agent框架	适配阿里云大模型，支持多模态交互（文本、图像）与MCP工具调用，文档完善	企业级应用开发（如智能客服、行业专属助手），需深度集成阿里云生态
Coze	字节跳动无代码AI Bot平台	可视化拖拽界面，内置知识库、多模态插件（如表情包、地图），零代码部署	快速搭建轻量级Bot（如社交平台机器人、个人事务助手），无需编程基础
Dify	开源LLM应用开发平台（API优先）	支持Prompt工程、灵活编排工具链，可私有化部署，注重开发者定制化	需深度定制的场景（如企业私有知识库助手、内部流程自动化工具）

三、智能体架构分类：三类经典设计与适用场景

Agent的架构决定其“决策方式”与“能力边界”，核心分为反应式（Reactive）、深思熟虑式（Deliberative）、混合式（Hybrid） 三类，三类架构各有侧重，覆盖从“快速响应”到“复杂规划”的全需求。

1. 反应式架构（Reactive）：“直觉型”快速响应

（1）核心特点

• 本质：基于“当前环境”即时决策，无长期规划，类似“膝跳反射”，依赖预设规则或LLM的即时判断。

• 核心逻辑：感知（获取环境输入）→ 决策（LLM/规则生成动作）→ 执行（反馈结果），循环直至任务完成。

（2）架构图与关键组件

Sensors（感知）→ Condition-Action Rules（条件-动作规则）→ Actuators（执行）
        ↑                          ↓
        └──────── Environment（环境反馈） ────────┘

• Sensors：获取环境输入（如用户问题、传感器数据）；

• Condition-Action Rules：核心决策层（如“用户问合格投资者标准→调用关键词搜索工具”）；

• Actuators：执行动作（如返回查询结果、触发警报）。

（3）优势与局限

优势	局限
速度快（毫秒级响应），无复杂推理	缺乏适应性，无法处理未预设场景
架构简单，易设计与验证	短视性，仅优化当前动作，可能陷入局部循环（如机器人绕圈）
可靠性高，行为由明确规则驱动	无法处理多步任务（如长期财务规划）

（4）典型应用场景

• 实时响应类：扫地机器人避障、无人机紧急悬停、高频交易触发；

• 简单交互类：游戏NPC对玩家攻击的即时反应、工业传感器超限警报；

• 规则明确类：FAQ问答、基础信息查询（如“私募基金合格投资者标准”）。

（5）实操案例：私募基金运作指引问答助手

案例目标

设计反应式Agent，自主回答用户关于私募基金规则的问题，具备“工具选择、知识边界感知、异常处理”能力。

核心实现步骤

1. 数据准备：构建私募基金规则知识库（FUND_RULES_DB），每条规则含ID、类别（如“设立与募集”）、问题、答案（如“合格投资者需净资产不低于1000万元”）；

2. 工具设计：开发3类核心工具，供Agent自主选择：

   • 关键词搜索工具（search_rules_by_keywords）：通过关键词匹配规则（如“合格投资者标准”）；

   • 类别查询工具（search_rules_by_category）：按规则类别检索（如“机构投资者资格”）；

   • 直接回答工具（answer_question）：基于问题与规则匹配度直接输出答案；

3. 架构搭建：基于LangChain Agent框架，定义3个核心模块：

   • 自定义提示模板（CustomPromptTemplate）：规范Agent思考格式（如“先判断问题类型→选择工具”）；

   • 自定义输出解析器（CustomOutputParser）：解析LLM输出，确定下一步动作（如“调用关键词搜索”）；

   • Agent执行器（AgentExecutor）：协调工具调用与结果返回；

4. 知识边界处理：识别超出知识库的问题（如“私募基金可投资资产类别”），明确区分“知识库内容”与“模型常识”，引导用户咨询专业渠道。

案例效果

• 当用户问“净资产800万元的机构能否投资私募基金”：

  1. Agent先尝试“类别查询（机构投资者资格）”，未找到结果；

  2. 自主切换“关键词搜索（机构投资者 净资产 资格）”，匹配到“合格投资者标准”规则；

  3. 输出结论：“机构需净资产≥1000万元，800万元不符合要求”，并提示额外条件（单只基金投资≥100万元）。

2. 深思熟虑式架构（Deliberative）：“战略型”复杂规划

（1）核心特点

• 本质：基于“内部世界模型”进行多步推理，优先优化长期目标，类似“下棋高手”，每步动作均经过推演。

• 核心逻辑：感知（收集数据）→ 建模（构建内部状态）→ 推理（生成候选方案）→ 决策（选择最优方案）→ 执行（反馈结果），注重全局最优。

（2）架构图与关键组件

Sensors（感知）→ World Model/Memory（内部模型/记忆）→ Planner/Reasoner（推理规划）→ Actuators（执行）
        ↑                          ↓                          ↓                          ↑
        └──────────────────────── Environment（环境反馈） ─────────────────────────┘

• World Model/Memory：构建环境的内部表示（如“当前市场处于衰退前期，锂价波动大”）；

• Planner/Reasoner：核心推理层，生成多个候选方案并模拟结果（如“投资组合调整的3种方案”）；

• 决策层：基于多维度指标（风险、收益、时间）选择最优方案。

（3）优势与局限

优势	局限
能处理多步复杂任务（如投研报告生成）	响应速度慢（需多轮推理），不适合实时场景
优化长期目标，避免短视决策	架构复杂，设计与验证成本高
适应动态环境（如市场变化）	依赖高质量内部模型，模型偏差会导致决策失误

（4）典型应用场景

• 战略规划类：物流路径优化、投资组合调整、长期财务规划；

• 复杂分析类：行业投研报告生成、风险评估与应对策略制定；

• 多目标优化类：平衡“收益-风险-流动性”的资产配置。

（5）实操案例：智能投研助手

案例目标

设计深思熟虑式Agent，基于LangGraph实现“新能源汽车行业中期投资机会”分析，生成结构化投研报告，包含市场背景、核心观点、风险评估、投资建议。

核心实现步骤

1. 状态定义：用ResearchAgentState类维护全流程状态，包含输入（研究主题、行业焦点、时间范围）、处理状态（感知数据、内部模型、候选方案）、输出（最终报告）；

2. 分阶段实现（基于LangGraph状态图）：

   阶段	核心任务	输出结果（结构化模型）
   感知（perception）	收集市场数据（销量、成本、政策）、行业趋势、近期新闻	PerceptionOutput：市场概况、关键指标（如“锂电池成本$107/kWh”）、近期新闻（如“特斯拉建墨西哥工厂”）
   建模（modeling）	分析市场状态、经济周期、风险因素（如锂价波动）、机会领域（中端电动车）	ModelingOutput：市场状态（“快速增长期”）、风险因素（“原材料价格波动”）、机会领域（“中端电动车+快充技术”）
   推理（reasoning）	生成3-5个候选投资方案，评估每个方案的假设、方法、预期结果、置信度	ReasoningPlan：方案ID、投资假设（“中端电动车销量年增30%”）、优缺点（如“成本低但竞争激烈”）
   决策（decision）	对比候选方案，选择最优解，形成投资论点与支持证据	DecisionOutput：选中方案ID、投资论点（“中端市场是核心增长极”）、风险评估、建议（“优先投资成本优化企业”）
   报告（report）	整合全流程结果，生成结构化报告（摘要、背景、观点、建议、时间框架）	最终投研报告：含市场数据、核心观点、风险因素、预期回报（“年化15%-20%”）

3. 错误处理：各阶段检查前置条件（如“推理阶段缺少内部模型→返回建模阶段”），确保流程不中断。

案例效果

生成的《新能源汽车行业投资研究报告》包含7大模块：

• 摘要：明确“中端电动车+下一代电池技术”为核心机会；

• 市场背景：全球电动车销量增42%、锂电池成本下降数据；

• 核心观点：中端市场3-5年销量年增30%-40%；

• 风险因素：原材料波动、技术迭代风险；

• 投资建议：优先配置具备规模效应的中端车企与电池厂商；

• 预期回报：中期年化15%-20%。

3. 混合式架构（Hybrid）：“平衡型”智能协同

（1）核心特点

• 本质：结合反应式的“快速响应”与深思熟虑式的“复杂规划”，通过“协调层”动态切换模式，兼顾“效率”与“智能”。

• 三层设计：

  1. 底层（反应式层）：处理紧急/简单任务（如“查上证指数”），毫秒级响应；

  2. 中层（协调层）：评估任务类型与优先级，选择处理模式（反应式/深思熟虑）；

  3. 顶层（深思熟虑层）：处理复杂任务（如“投资组合优化”），多步推理。

（2）架构图与关键组件

Sensors（感知）→ 协调层（任务评估）→ 反应式层（紧急/简单任务）→ Actuators（执行）
                          ↓
                          → 深思熟虑层（复杂任务：建模→推理→决策）→ Actuators（执行）
                          ↓
                          ← Environment（环境反馈）

• 协调层：核心“调度中心”，基于“任务复杂性、时效性、资源需求”选择模式（如“查账户余额→反应式，做退休规划→深思熟虑”）；

• 仲裁机制：紧急情况（如“账户异常交易”）强制启用反应式，确保快速响应。

（3）优势与典型场景

• 核心优势：

  • 实时性：紧急任务毫秒级响应；

  • 智能性：复杂任务深度规划；

  • 灵活性：动态适配任务需求，避免“一刀切”。

• 典型场景：

  • 自动驾驶：突发障碍→反应式刹车，正常行驶→深思熟虑规划路线；

  • 智能投顾：查市场行情→反应式，做退休财务规划→深思熟虑；

  • 企业客服：基础问题（“营业时间”）→反应式，复杂投诉（“理赔纠纷”）→深思熟虑。

（4）实操案例：投顾AI助手

案例目标

设计混合式Agent，为“平衡型投资者”提供财富管理服务，既能快速回答市场信息，又能深度分析投资组合调整策略。

核心实现步骤

1. 状态定义：用WealthAdvisorState类维护状态，包含输入（用户查询、客户画像）、处理状态（查询类型、处理模式、市场数据）、输出（最终响应）；

2. 协调层决策逻辑：基于3个维度评估任务：

   评估维度	反应式触发条件	深思熟虑触发条件
   任务复杂性	简单（单轮查询，无多步推理）	复杂（多步分析，需建模与方案生成）
   时效性需求	高（需实时数据，如“今天上证指数”）	低（允许长耗时，如“退休规划”）
   资源需求	低（仅需基础数据，无复杂计算）	高（需整合市场数据、客户画像、历史业绩）

3. 分模式处理流程：

   • 反应式流程（如“查上证指数”）：协调层判断→调用市场数据接口→快速返回结果（“收于3200.12点，涨0.49%”），耗时约11秒；

   • 深思熟虑流程（如“应对经济衰退的组合调整”）：协调层判断→数据收集（市场趋势、客户风险承受力）→深度分析（构建衰退模型、生成3种调整方案）→决策（选择“防御性增长”方案）→生成建议（股票降10%、债券升10%、现金升5%）→输出结构化报告，耗时约112秒。

案例效果

• 对“平衡型投资者”的150万元组合，给出调整建议：

  资产类别	当前比例	调整后比例	调整后金额（元）	调整逻辑
  股票	40%	30%	450,000	降低波动，优先配置医疗、消费品等防御性行业
  债券	30%	40%	600,000	提升稳定性，配置短期政府债/投资级企业债
  现金	10%	15%	225,000	增加流动性，用于应急或抄底
  另类投资	20%	15%	225,000	降低风险，配置低相关性的私募股权/基础设施基金

四、构建Agent的核心思想：三大原则与实践建议

文档在最后提炼了“避免踩坑”的核心方法论，强调“不盲目追求复杂，只做适配需求的设计”，具体包括三大原则：

1. 原则1：不要为所有任务构建Agent

Agent适合复杂、模糊、高价值的任务，而非“一刀切”应用，判断标准需满足4点：

• 任务复杂性：若能通过明确决策树规划（如“根据温度推荐衣服”），直接用工作流更高效，无需Agent；

• 任务价值：Agent的Token消耗与开发成本高，需由高回报任务承担（如代码生成、投研分析），低价值任务（如“计算1+1”）无需Agent；

• 关键能力验证：确保Agent能处理核心子任务（如代码Agent需能生成+调试代码），避免“看似智能，实则无法落地”；

• 错误成本：高风险场景（如医疗诊断、大额交易）需限制Agent权限，必要时加入人工审核，避免错误累积。

案例：代码生成是理想场景——复杂性高（需理解需求、语法、逻辑）、价值大（节省开发时间）、输出易验证（通过单元测试）。

2. 原则2：保持Agent设计的简洁性

Agent的核心竞争力在于“解决问题”，而非“架构复杂”，设计时需聚焦3个核心组件，避免过度抽象：

• 环境（Environment）：Agent的“操作系统”，明确Agent能感知的输入（如用户问题、工具反馈）与能执行的动作（如调用API、生成报告）；

• 工具集（Tools）：提供“行动接口”，工具需满足“功能明确、文档清晰、调用稳定”（如“关键词搜索工具”需明确输入输出格式）；

• 系统提示（System Prompt）：定义Agent的目标、约束与行为准则（如“作为投顾，需优先提示风险，不承诺收益”）。

实践建议：初期用最小原型验证核心功能（如仅保留“关键词搜索+直接回答”工具），待基础行为稳定后，再优化成本、延迟等细节；不同功能的Agent可共享代码框架，仅调整工具与提示。

3. 原则3：像Agent一样思考，理解其局限性

Agent的决策基于“有限上下文”与“工具能力”，设计时需模拟Agent视角，发现潜在缺陷：

• 理解上下文限制：LLM的上下文窗口有限（如10-20k Token），Agent无法记住超长历史（如用户1年前的对话），需通过长期记忆工具补充；

• 模拟操作视角：若Agent需操作电脑（如生成文件），需考虑其“感知边界”（如仅能通过静态截图获取界面信息，无法像人类一样直观操作）；

• 优化反馈机制：通过“直接询问模型”验证设计（如“这个工具调用指令是否清晰？”），或分析轨迹日志（如Agent反复调用同一工具却失败），调整提示或工具设计。

核心目标：在“能力提升”与“成本、风险、用户体验”之间找平衡，避免为了“智能”而牺牲“可靠性”。

五、总结：Agent智能体的核心价值与应用方向

本文通过“定义-架构-案例-方法论”的逻辑，清晰传递了Agent的核心价值：

• 从“工具”到“伙伴”：Agent不再是“被动执行指令的工具”，而是“主动理解目标、规划步骤、解决问题的智能伙伴”；

• 三类架构覆盖全需求：反应式应对“实时简单任务”，深思熟虑式处理“复杂规划任务”，混合式平衡“效率与智能”，可根据场景灵活选择；

• 构建需务实：不盲目追求复杂架构，聚焦“环境、工具、提示”三大核心，优先验证核心能力，再逐步优化。

未来Agent的应用将更聚焦“行业深度”（如金融投研、医疗诊断、智能制造），通过“多Agent协作”与“跨领域工具集成”，解决更复杂的现实问题。

Langchain sample code

智能投研助手：基于LangGraph的深思熟虑型AI Agent，让投资决策更智能！

在金融投资领域，做出明智的投资决策需要深入的市场分析、全面的数据收集和严谨的逻辑推理。传统的投资分析往往依赖于人工研究，耗时耗力且容易受到主观因素影响。而现在，借助AI Agent技术，我们可以构建一个"深思熟虑"的智能投研助手，它能够像专业的投资分析师一样，系统性地进行市场研究并生成专业的投资报告。

今天，我们就来深入解析一个基于LangGraph实现的深思熟虑型智能投研助手，看看它是如何通过五个阶段的深度分析，帮助我们做出更明智的投资决策！

一、什么是深思熟虑型Agent？

在AI Agent的分类中，深思熟虑型（Deliberative）Agent是最接近人类专家思维模式的一种。它不像反应式Agent那样快速响应，而是通过以下五个阶段进行深度思考：

1. 感知（Perception）：收集市场数据和信息

2. 建模（Modeling）：构建内部世界模型，理解市场状态

3. 推理（Reasoning）：生成多个候选分析方案并模拟结果

4. 决策（Decision）：选择最优投资观点

5. 报告（Report）：生成完整研究报告

这种设计使得Agent能够像专业的投资分析师一样，进行全面而深入的分析。

二、核心代码解析

让我们深入解析这个智能投研助手的核心实现代码：

2.1 数据模型定义

首先，我们定义了各个阶段的输出数据模型：

class PerceptionOutput(BaseModel):
    """感知阶段输出的市场数据和信息"""
    market_overview: str = Field(..., description="市场概况和最新动态")
    key_indicators: Dict[str, str] = Field(..., description="关键经济和市场指标")
    recent_news: List[str] = Field(..., description="近期重要新闻")
    industry_trends: Dict[str, str] = Field(..., description="行业趋势分析")
​
class ModelingOutput(BaseModel):
    """建模阶段输出的内部世界模型"""
    market_state: str = Field(..., description="当前市场状态评估")
    economic_cycle: str = Field(..., description="经济周期判断")
    risk_factors: List[str] = Field(..., description="主要风险因素")
    opportunity_areas: List[str] = Field(..., description="潜在机会领域")
    market_sentiment: str = Field(..., description="市场情绪分析")

这些模型确保了每个阶段的输出都有明确的结构和含义。

2.2 Agent状态管理

通过TypedDict定义Agent的状态，确保整个分析过程的状态可追踪：

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]  # 错误信息

2.3 五个核心阶段实现

感知阶段：收集市场数据

def perception(state: ResearchAgentState) -> ResearchAgentState:
    """感知阶段：收集和整理市场数据和信息"""
    print("1. 感知阶段：收集市场数据和信息...")
    
    try:
        # 准备提示
        prompt = ChatPromptTemplate.from_template(PERCEPTION_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"]
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "perception_data": result,
            "current_phase": "modeling"
        }
    except Exception as e:
        return {
            **state,
            "error": f"感知阶段出错: {str(e)}",
            "current_phase": "perception"  # 保持在当前阶段
        }

建模阶段：构建内部世界模型

def modeling(state: ResearchAgentState) -> ResearchAgentState:
    """建模阶段：构建内部世界模型，理解市场状态"""
    print("2. 建模阶段：构建内部世界模型...")
    
    try:
        # 确保感知数据已存在
        if not state.get("perception_data"):
            return {
                **state,
                "error": "建模阶段缺少感知数据",
                "current_phase": "perception"  # 回到感知阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(MODELING_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "world_model": result,
            "current_phase": "reasoning"
        }
    except Exception as e:
        return {
            **state,
            "error": f"建模阶段出错: {str(e)}",
            "current_phase": "modeling"  # 保持在当前阶段
        }

推理阶段：生成候选分析方案

def reasoning(state: ResearchAgentState) -> ResearchAgentState:
    """推理阶段：生成多个候选分析方案并模拟结果"""
    print("3. 推理阶段：生成候选分析方案...")
    
    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)
        }
        
        # 调用LLM
        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"  # 保持在当前阶段
        }

决策阶段：选择最优方案

def decision(state: ResearchAgentState) -> ResearchAgentState:
    """决策阶段：评估候选方案并选择最优投资观点"""
    print("4. 决策阶段：选择最优投资观点...")
    
    try:
        # 确保候选方案已存在
        if not state.get("reasoning_plans"):
            return {
                **state,
                "error": "决策阶段缺少候选方案",
                "current_phase": "reasoning"  # 回到推理阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(DECISION_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),
            "reasoning_plans": json.dumps(state["reasoning_plans"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "selected_plan": result,
            "current_phase": "report"
        }
    except Exception as e:
        return {
            **state,
            "error": f"决策阶段出错: {str(e)}",
            "current_phase": "decision"  # 保持在当前阶段
        }

报告阶段：生成完整研究报告

def report_generation(state: ResearchAgentState) -> ResearchAgentState:
    """报告阶段：生成完整的投资研究报告"""
    print("5. 报告阶段：生成完整研究报告...")
    
    try:
        # 确保选定方案已存在
        if not state.get("selected_plan"):
            return {
                **state,
                "error": "报告阶段缺少选定方案",
                "current_phase": "decision"  # 回到决策阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(REPORT_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2),
            "world_model": json.dumps(state["world_model"], ensure_ascii=False, indent=2),
            "selected_plan": json.dumps(state["selected_plan"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | StrOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "final_report": result,
            "current_phase": "completed"
        }
    except Exception as e:
        return {
            **state,
            "error": f"报告生成阶段出错: {str(e)}",
            "current_phase": "report"  # 保持在当前阶段
        }

2.4 工作流编排

使用LangGraph将五个阶段串联起来：

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()

三、实际运行效果展示

当运行这个智能投研助手时，你会看到以下的执行过程：

=== 深思熟虑智能体 - 智能投研助手 ===
​
使用模型：Qwen-Turbo-2025-04-28
​
请输入研究主题 (例如: 新能源汽车行业投资机会): 新能源汽车行业投资机会
请输入行业焦点 (例如: 电动汽车制造、电池技术): 电动汽车制造
请输入时间范围 [短期/中期/长期]: 中期
​
智能投研助手开始工作...
​
1. 感知阶段：收集市场数据和信息...
2. 建模阶段：构建内部世界模型...
3. 推理阶段：生成候选分析方案...
4. 决策阶段：选择最优投资观点...
5. 报告阶段：生成完整研究报告...
​
=== 最终研究报告 ===
​
# 新能源汽车行业投资研究报告
​
## 摘要
本报告分析了新能源汽车行业的中期投资机会，重点关注电动汽车制造领域。基于当前市场数据和趋势分析，我们认为该行业在未来3-5年内仍将保持高速增长，但竞争将日趋激烈。建议投资者重点关注具备技术优势和规模效应的企业。
​
## 市场和行业背景
...（详细分析内容）...
​
## 核心投资观点
...（投资观点和论据）...
​
## 详细分析论证
...（深入的分析和论证）...
​
## 风险因素
...（风险评估）...
​
## 投资建议
...（具体的投资建议）...
​
## 时间框架和预期回报
建议投资时间框架为3-5年，预期年化回报率约为15%-25%。
​
报告已保存为: research_report_20250823_143022.txt

四、未来应用场景展望

这个智能投研助手的架构具有很强的扩展性和适应性，可以应用于多个场景：

4.1 个人投资顾问

• 为个人投资者提供专业的投资建议

• 根据个人风险偏好定制投资策略

• 实时监控投资组合并提供调整建议

4.2 机构投研支持

• 辅助专业分析师进行行业研究

• 自动生成初步研究报告，提高工作效率

• 提供多角度的分析思路和方案

4.3 教育培训工具

• 帮助金融专业学生理解投资分析流程

• 提供标准化的投资分析框架训练

• 模拟不同市场环境下的投资决策

4.4 风险管理

• 识别潜在的投资风险因素

• 提供风险预警和应对策略

• 评估投资组合的风险水平

五、技术优势总结

5.1 系统性分析

通过五个阶段的深度分析，确保不遗漏任何重要信息和因素。

5.2 结构化输出

使用Pydantic模型确保输出的结构化和一致性，便于后续处理和分析。

5.3 错误处理机制

完善的错误处理机制确保系统在遇到问题时能够优雅地处理并继续执行。

5.4 可视化工作流

通过LangGraph提供的可视化功能，可以清晰地看到整个分析过程的流程图。

六、结语

这个基于LangGraph的深思熟虑型智能投研助手展示了AI Agent在金融投资领域的巨大潜力。它不仅能够自动化地完成复杂的投资分析工作，还能提供结构化的分析思路和专业的投资建议。

随着AI技术的不断发展和完善，我们可以预见，这样的智能投研助手将在未来的投资领域发挥越来越重要的作用，帮助投资者做出更加明智和科学的投资决策。

无论是专业的投资机构还是个人投资者，都可以从这样的AI助手工具中受益。它不仅能够提高投资分析的效率，还能帮助我们避免一些常见的认知偏差，让投资决策更加理性和科学。

未来，随着更多数据源的接入和更先进AI模型的应用，这样的智能投研助手将变得更加智能和强大，成为我们投资路上的得力助手！

---

完整可运行代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
​
"""
深思熟虑智能体（Deliberative Agent）- 智能投研助手
​
基于LangGraph实现的深思熟虑型智能体，适用于投资研究场景，能够整合数据，
进行多步骤分析和推理，生成投资观点和研究报告。
​
核心流程：
1. 感知：收集市场数据和信息
2. 建模：构建内部世界模型，理解市场状态
3. 推理：生成多个候选分析方案并模拟结果
4. 决策：选择最优投资观点并形成报告
5. 报告：生成完整研究报告
"""
​
import os
import json
from typing import Dict, List, Any, Literal, TypedDict, Optional, Union, Tuple
from datetime import datetime
​
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.llms import Tongyi
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from langgraph.graph import StateGraph, END
​
# 设置API密钥
DASHSCOPE_API_KEY = 'sk-882e296067b744289acf27e6e20f3ec0'
​
# 创建LLM实例
llm = Tongyi(model_name="Qwen-Turbo-2025-04-28", dashscope_api_key=DASHSCOPE_API_KEY)
​
# 定义输出模型
class PerceptionOutput(BaseModel):
    """感知阶段输出的市场数据和信息"""
    market_overview: str = Field(..., description="市场概况和最新动态")
    key_indicators: Dict[str, str] = Field(..., description="关键经济和市场指标")
    recent_news: List[str] = Field(..., description="近期重要新闻")
    industry_trends: Dict[str, str] = Field(..., description="行业趋势分析")
​
class ModelingOutput(BaseModel):
    """建模阶段输出的内部世界模型"""
    market_state: str = Field(..., description="当前市场状态评估")
    economic_cycle: str = Field(..., description="经济周期判断")
    risk_factors: List[str] = Field(..., description="主要风险因素")
    opportunity_areas: List[str] = Field(..., description="潜在机会领域")
    market_sentiment: str = Field(..., description="市场情绪分析")
​
class ReasoningPlan(BaseModel):
    """推理阶段生成的候选分析方案"""
    plan_id: str = Field(..., description="方案ID")
    hypothesis: str = Field(..., description="投资假设")
    analysis_approach: str = Field(..., description="分析方法")
    expected_outcome: str = Field(..., description="预期结果")
    confidence_level: float = Field(..., description="置信度(0-1)")
    pros: List[str] = Field(..., description="方案优势")
    cons: List[str] = Field(..., description="方案劣势")
​
class DecisionOutput(BaseModel):
    """决策阶段选择的最优投资观点"""
    selected_plan_id: str = Field(..., description="选中的方案ID")
    investment_thesis: str = Field(..., description="投资论点")
    supporting_evidence: List[str] = Field(..., description="支持证据")
    risk_assessment: str = Field(..., description="风险评估")
    recommendation: str = Field(..., description="投资建议")
    timeframe: str = Field(..., description="时间框架")
    
# 定义智能体状态
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]  # 错误信息
​
​
# 提示模板
PERCEPTION_PROMPT = """你是一个专业的投资研究分析师，请收集和整理关于以下研究主题的市场数据和信息：
​
研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}
​
请从以下几个方面进行市场感知：
1. 市场概况和最新动态
2. 关键经济和市场指标
3. 近期重要新闻（至少3条）
4. 行业趋势分析（至少针对3个细分领域）
​
根据你的专业知识和经验，提供尽可能详细和准确的信息。
​
输出格式要求为JSON，包含以下字段：
- market_overview: 字符串
- key_indicators: 字典，键为指标名称，值为指标值和简要解释
- recent_news: 字符串列表，每项为一条重要新闻
- industry_trends: 字典，键为细分领域，值为趋势分析
"""
​
MODELING_PROMPT = """你是一个资深投资策略师，请根据以下市场数据和信息，构建市场内部模型，进行深度分析：
​
研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}
​
市场数据和信息:
{perception_data}
​
请构建一个全面的市场内部模型，包括：
1. 当前市场状态评估
2. 经济周期判断
3. 主要风险因素（至少3个）
4. 潜在机会领域（至少3个）
5. 市场情绪分析
​
输出格式要求为JSON，包含以下字段：
- market_state: 字符串
- economic_cycle: 字符串
- risk_factors: 字符串列表
- opportunity_areas: 字符串列表
- market_sentiment: 字符串
"""
​
REASONING_PROMPT = """你是一个战略投资顾问，请根据以下市场模型，生成3个不同的投资分析方案：
​
研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}
​
市场内部模型:
{world_model}
​
请为每个方案提供：
1. 方案ID（简短标识符）
2. 投资假设
3. 分析方法
4. 预期结果
5. 置信度（0-1之间的小数）
6. 方案优势（至少3点）
7. 方案劣势（至少2点）
​
这些方案应该有明显的差异，代表不同的投资思路或分析角度。
​
输出格式要求为JSON数组，每个元素包含以下字段：
- plan_id: 字符串
- hypothesis: 字符串
- analysis_approach: 字符串
- expected_outcome: 字符串
- confidence_level: 浮点数
- pros: 字符串列表
- cons: 字符串列表
"""
​
DECISION_PROMPT = """你是一个投资决策委员会主席，请评估以下候选分析方案，选择最优方案并形成投资决策：
​
研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}
​
市场内部模型:
{world_model}
​
候选分析方案:
{reasoning_plans}
​
请基于方案的假设、分析方法、预期结果、置信度以及优缺点，选择最优的投资方案，并给出详细的决策理由。
你的决策应该综合考虑投资潜力、风险水平和时间框架的匹配度。
​
输出格式要求为JSON，包含以下字段：
- selected_plan_id: 字符串
- investment_thesis: 字符串
- supporting_evidence: 字符串列表
- risk_assessment: 字符串
- recommendation: 字符串
- timeframe: 字符串
"""
​
REPORT_PROMPT = """你是一个专业的投资研究报告撰写人，请根据以下信息生成一份完整的投资研究报告：
​
研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}
​
市场数据和信息:
{perception_data}
​
市场内部模型:
{world_model}
​
选定的投资决策:
{selected_plan}
​
请生成一份结构完整、逻辑清晰的投研报告，包括但不限于：
1. 报告标题和摘要
2. 市场和行业背景
3. 核心投资观点
4. 详细分析论证
5. 风险因素
6. 投资建议
7. 时间框架和预期回报
​
报告应当专业、客观，同时提供足够的分析深度和洞见。
"""
​
# 第一阶段：感知 - 收集市场数据和信息
def perception(state: ResearchAgentState) -> ResearchAgentState:
    """感知阶段：收集和整理市场数据和信息"""
    
    print("1. 感知阶段：收集市场数据和信息...")
    
    try:
        # 准备提示
        prompt = ChatPromptTemplate.from_template(PERCEPTION_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"]
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "perception_data": result,
            "current_phase": "modeling"
        }
    except Exception as e:
        return {
            **state,
            "error": f"感知阶段出错: {str(e)}",
            "current_phase": "perception"  # 保持在当前阶段
        }
​
# 第二阶段：建模 - 构建内部世界模型
def modeling(state: ResearchAgentState) -> ResearchAgentState:
    """建模阶段：构建内部世界模型，理解市场状态"""
    
    print("2. 建模阶段：构建内部世界模型...")
    
    try:
        # 确保感知数据已存在
        if not state.get("perception_data"):
            return {
                **state,
                "error": "建模阶段缺少感知数据",
                "current_phase": "perception"  # 回到感知阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(MODELING_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "world_model": result,
            "current_phase": "reasoning"
        }
    except Exception as e:
        return {
            **state,
            "error": f"建模阶段出错: {str(e)}",
            "current_phase": "modeling"  # 保持在当前阶段
        }
​
# 第三阶段：推理 - 生成候选分析方案
def reasoning(state: ResearchAgentState) -> ResearchAgentState:
    """推理阶段：生成多个候选分析方案并模拟结果"""
    
    print("3. 推理阶段：生成候选分析方案...")
    
    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)
        }
        
        # 调用LLM
        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"  # 保持在当前阶段
        }
​
# 第四阶段：决策 - 选择最优方案
def decision(state: ResearchAgentState) -> ResearchAgentState:
    """决策阶段：评估候选方案并选择最优投资观点"""
    
    print("4. 决策阶段：选择最优投资观点...")
    
    try:
        # 确保候选方案已存在
        if not state.get("reasoning_plans"):
            return {
                **state,
                "error": "决策阶段缺少候选方案",
                "current_phase": "reasoning"  # 回到推理阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(DECISION_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),
            "reasoning_plans": json.dumps(state["reasoning_plans"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "selected_plan": result,
            "current_phase": "report"
        }
    except Exception as e:
        return {
            **state,
            "error": f"决策阶段出错: {str(e)}",
            "current_phase": "decision"  # 保持在当前阶段
        }
​
# 第五阶段：报告 - 生成完整研究报告
def report_generation(state: ResearchAgentState) -> ResearchAgentState:
    """报告阶段：生成完整的投资研究报告"""
    
    print("5. 报告阶段：生成完整研究报告...")
    
    try:
        # 确保选定方案已存在
        if not state.get("selected_plan"):
            return {
                **state,
                "error": "报告阶段缺少选定方案",
                "current_phase": "decision"  # 回到决策阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(REPORT_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2),
            "world_model": json.dumps(state["world_model"], ensure_ascii=False, indent=2),
            "selected_plan": json.dumps(state["selected_plan"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | StrOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "final_report": result,
            "current_phase": "completed"
        }
    except Exception as e:
        return {
            **state,
            "error": f"报告生成阶段出错: {str(e)}",
            "current_phase": "report"  # 保持在当前阶段
        }
​
# 路由函数 - 根据当前阶段决定下一步
def router(state: ResearchAgentState) -> str:
    """根据当前阶段路由到下一步或结束"""
    
    # 如果有错误，保持在当前阶段
    if state.get("error"):
        return state["current_phase"]
    
    # 根据当前阶段决定下一步
    current = state["current_phase"]
    
    if current == "perception":
        return "modeling"
    elif current == "modeling":
        return "reasoning"
    elif current == "reasoning":
        return "decision"
    elif current == "decision":
        return "report"
    elif current == "report":
        return END
    else:
        return END
​
# 创建智能体工作流图
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()
​
# 测试函数
def run_research_agent(topic: str, industry: str, horizon: str) -> Dict[str, Any]:
    """运行研究智能体并返回结果"""
    
    # 创建工作流
    agent = create_research_agent_workflow()
    
    # 准备初始状态
    initial_state = {
        "research_topic": topic,
        "industry_focus": industry,
        "time_horizon": horizon,
        "perception_data": None,
        "world_model": None,
        "reasoning_plans": None,
        "selected_plan": None,
        "final_report": None,
        "current_phase": "perception",
        "error": None
    }
    print("LangGraph Mermaid流程图：")
    print(agent.get_graph().draw_mermaid())
    
    # 运行智能体
    result = agent.invoke(initial_state)
    
    return result
​
# 主函数
if __name__ == "__main__":
    print("=== 深思熟虑智能体 - 智能投研助手 ===\n")
    print("使用模型：Qwen-Turbo-2025-04-28\n")
    
    # 用户输入
    topic = input("请输入研究主题 (例如: 新能源汽车行业投资机会): ")
    industry = input("请输入行业焦点 (例如: 电动汽车制造、电池技术): ")
    horizon = input("请输入时间范围 [短期/中期/长期]: ")
    
    print("\n智能投研助手开始工作...\n")
    
    try:
        # 运行智能体
        result = run_research_agent(topic, industry, horizon)
        
        # 处理结果
        if result.get("error"):
            print(f"\n发生错误: {result['error']}")
        else:
            print("\n=== 最终研究报告 ===\n")
            print(result.get("final_report", "未生成报告"))
            
            # 保存报告
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"research_report_{timestamp}.txt"
            
            with open(filename, "w", encoding="utf-8") as f:
                f.write(result.get("final_report", "未生成报告"))
            
            print(f"\n报告已保存为: {filename}")
            
    except Exception as e:
        print(f"\n运行过程中发生错误: {str(e)}")

运行步骤

1. 首先安装依赖：

pip install langchain langgraph dashscope

1. 设置环境变量（或在代码中直接设置API密钥）：

export DASHSCOPE_API_KEY=你的API密钥

1. 运行代码：

python deliberative_research_langgraph.py

现在，你已经掌握了如何构建一个深思熟虑型的智能投研助手！快去尝试运行并体验它的强大功能吧！

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
深思熟虑智能体（Deliberative Agent）- 智能投研助手

基于LangGraph实现的深思熟虑型智能体，适用于投资研究场景，能够整合数据，
进行多步骤分析和推理，生成投资观点和研究报告。

核心流程：
1. 感知：收集市场数据和信息
2. 建模：构建内部世界模型，理解市场状态
3. 推理：生成多个候选分析方案并模拟结果
4. 决策：选择最优投资观点并形成报告
5. 报告：生成完整研究报告
"""

import os
import json
from typing import Dict, List, Any, Literal, TypedDict, Optional, Union, Tuple
from datetime import datetime

from langchain_core.prompts import ChatPromptTemplate
from langchain_community.llms import Tongyi
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field, validator
from langgraph.graph import StateGraph, END

# 设置API密钥
DASHSCOPE_API_KEY = 'sk-882e296067b744289acf27e6e20f3ec0'

# 创建LLM实例
llm = Tongyi(model_name="Qwen-Turbo-2025-04-28", dashscope_api_key=DASHSCOPE_API_KEY)

# 定义输出模型
class PerceptionOutput(BaseModel):
    """感知阶段输出的市场数据和信息"""
    market_overview: str = Field(..., description="市场概况和最新动态")
    key_indicators: Dict[str, str] = Field(..., description="关键经济和市场指标")
    recent_news: List[str] = Field(..., description="近期重要新闻")
    industry_trends: Dict[str, str] = Field(..., description="行业趋势分析")

class ModelingOutput(BaseModel):
    """建模阶段输出的内部世界模型"""
    market_state: str = Field(..., description="当前市场状态评估")
    economic_cycle: str = Field(..., description="经济周期判断")
    risk_factors: List[str] = Field(..., description="主要风险因素")
    opportunity_areas: List[str] = Field(..., description="潜在机会领域")
    market_sentiment: str = Field(..., description="市场情绪分析")

class ReasoningPlan(BaseModel):
    """推理阶段生成的候选分析方案"""
    plan_id: str = Field(..., description="方案ID")
    hypothesis: str = Field(..., description="投资假设")
    analysis_approach: str = Field(..., description="分析方法")
    expected_outcome: str = Field(..., description="预期结果")
    confidence_level: float = Field(..., description="置信度(0-1)")
    pros: List[str] = Field(..., description="方案优势")
    cons: List[str] = Field(..., description="方案劣势")

class DecisionOutput(BaseModel):
    """决策阶段选择的最优投资观点"""
    selected_plan_id: str = Field(..., description="选中的方案ID")
    investment_thesis: str = Field(..., description="投资论点")
    supporting_evidence: List[str] = Field(..., description="支持证据")
    risk_assessment: str = Field(..., description="风险评估")
    recommendation: str = Field(..., description="投资建议")
    timeframe: str = Field(..., description="时间框架")
    
# 定义智能体状态
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]  # 错误信息


# 提示模板
PERCEPTION_PROMPT = """你是一个专业的投资研究分析师，请收集和整理关于以下研究主题的市场数据和信息：

研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}

请从以下几个方面进行市场感知：
1. 市场概况和最新动态
2. 关键经济和市场指标
3. 近期重要新闻（至少3条）
4. 行业趋势分析（至少针对3个细分领域）

根据你的专业知识和经验，提供尽可能详细和准确的信息。

输出格式要求为JSON，包含以下字段：
- market_overview: 字符串
- key_indicators: 字典，键为指标名称，值为指标值和简要解释
- recent_news: 字符串列表，每项为一条重要新闻
- industry_trends: 字典，键为细分领域，值为趋势分析
"""

MODELING_PROMPT = """你是一个资深投资策略师，请根据以下市场数据和信息，构建市场内部模型，进行深度分析：

研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}

市场数据和信息:
{perception_data}

请构建一个全面的市场内部模型，包括：
1. 当前市场状态评估
2. 经济周期判断
3. 主要风险因素（至少3个）
4. 潜在机会领域（至少3个）
5. 市场情绪分析

输出格式要求为JSON，包含以下字段：
- market_state: 字符串
- economic_cycle: 字符串
- risk_factors: 字符串列表
- opportunity_areas: 字符串列表
- market_sentiment: 字符串
"""

REASONING_PROMPT = """你是一个战略投资顾问，请根据以下市场模型，生成3个不同的投资分析方案：

研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}

市场内部模型:
{world_model}

请为每个方案提供：
1. 方案ID（简短标识符）
2. 投资假设
3. 分析方法
4. 预期结果
5. 置信度（0-1之间的小数）
6. 方案优势（至少3点）
7. 方案劣势（至少2点）

这些方案应该有明显的差异，代表不同的投资思路或分析角度。

输出格式要求为JSON数组，每个元素包含以下字段：
- plan_id: 字符串
- hypothesis: 字符串
- analysis_approach: 字符串
- expected_outcome: 字符串
- confidence_level: 浮点数
- pros: 字符串列表
- cons: 字符串列表
"""

DECISION_PROMPT = """你是一个投资决策委员会主席，请评估以下候选分析方案，选择最优方案并形成投资决策：

研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}

市场内部模型:
{world_model}

候选分析方案:
{reasoning_plans}

请基于方案的假设、分析方法、预期结果、置信度以及优缺点，选择最优的投资方案，并给出详细的决策理由。
你的决策应该综合考虑投资潜力、风险水平和时间框架的匹配度。

输出格式要求为JSON，包含以下字段：
- selected_plan_id: 字符串
- investment_thesis: 字符串
- supporting_evidence: 字符串列表
- risk_assessment: 字符串
- recommendation: 字符串
- timeframe: 字符串
"""

REPORT_PROMPT = """你是一个专业的投资研究报告撰写人，请根据以下信息生成一份完整的投资研究报告：

研究主题: {research_topic}
行业焦点: {industry_focus}
时间范围: {time_horizon}

市场数据和信息:
{perception_data}

市场内部模型:
{world_model}

选定的投资决策:
{selected_plan}

请生成一份结构完整、逻辑清晰的投研报告，包括但不限于：
1. 报告标题和摘要
2. 市场和行业背景
3. 核心投资观点
4. 详细分析论证
5. 风险因素
6. 投资建议
7. 时间框架和预期回报

报告应当专业、客观，同时提供足够的分析深度和洞见。
"""

# 第一阶段：感知 - 收集市场数据和信息
def perception(state: ResearchAgentState) -> ResearchAgentState:
    """感知阶段：收集和整理市场数据和信息"""
    
    print("1. 感知阶段：收集市场数据和信息...")
    
    try:
        # 准备提示
        prompt = ChatPromptTemplate.from_template(PERCEPTION_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"]
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "perception_data": result,
            "current_phase": "modeling"
        }
    except Exception as e:
        return {
            **state,
            "error": f"感知阶段出错: {str(e)}",
            "current_phase": "perception"  # 保持在当前阶段
        }

# 第二阶段：建模 - 构建内部世界模型
def modeling(state: ResearchAgentState) -> ResearchAgentState:
    """建模阶段：构建内部世界模型，理解市场状态"""
    
    print("2. 建模阶段：构建内部世界模型...")
    
    try:
        # 确保感知数据已存在
        if not state.get("perception_data"):
            return {
                **state,
                "error": "建模阶段缺少感知数据",
                "current_phase": "perception"  # 回到感知阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(MODELING_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "world_model": result,
            "current_phase": "reasoning"
        }
    except Exception as e:
        return {
            **state,
            "error": f"建模阶段出错: {str(e)}",
            "current_phase": "modeling"  # 保持在当前阶段
        }

# 第三阶段：推理 - 生成候选分析方案
def reasoning(state: ResearchAgentState) -> ResearchAgentState:
    """推理阶段：生成多个候选分析方案并模拟结果"""
    
    print("3. 推理阶段：生成候选分析方案...")
    
    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)
        }
        
        # 调用LLM
        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"  # 保持在当前阶段
        }

# 第四阶段：决策 - 选择最优方案
def decision(state: ResearchAgentState) -> ResearchAgentState:
    """决策阶段：评估候选方案并选择最优投资观点"""
    
    print("4. 决策阶段：选择最优投资观点...")
    
    try:
        # 确保候选方案已存在
        if not state.get("reasoning_plans"):
            return {
                **state,
                "error": "决策阶段缺少候选方案",
                "current_phase": "reasoning"  # 回到推理阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(DECISION_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),
            "reasoning_plans": json.dumps(state["reasoning_plans"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | JsonOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "selected_plan": result,
            "current_phase": "report"
        }
    except Exception as e:
        return {
            **state,
            "error": f"决策阶段出错: {str(e)}",
            "current_phase": "decision"  # 保持在当前阶段
        }

# 第五阶段：报告 - 生成完整研究报告
def report_generation(state: ResearchAgentState) -> ResearchAgentState:
    """报告阶段：生成完整的投资研究报告"""
    
    print("5. 报告阶段：生成完整研究报告...")
    
    try:
        # 确保选定方案已存在
        if not state.get("selected_plan"):
            return {
                **state,
                "error": "报告阶段缺少选定方案",
                "current_phase": "decision"  # 回到决策阶段
            }
        
        # 准备提示
        prompt = ChatPromptTemplate.from_template(REPORT_PROMPT)
        
        # 构建输入
        input_data = {
            "research_topic": state["research_topic"],
            "industry_focus": state["industry_focus"],
            "time_horizon": state["time_horizon"],
            "perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2),
            "world_model": json.dumps(state["world_model"], ensure_ascii=False, indent=2),
            "selected_plan": json.dumps(state["selected_plan"], ensure_ascii=False, indent=2)
        }
        
        # 调用LLM
        chain = prompt | llm | StrOutputParser()
        result = chain.invoke(input_data)
        
        # 更新状态
        return {
            **state,
            "final_report": result,
            "current_phase": "completed"
        }
    except Exception as e:
        return {
            **state,
            "error": f"报告生成阶段出错: {str(e)}",
            "current_phase": "report"  # 保持在当前阶段
        }

# 路由函数 - 根据当前阶段决定下一步
def router(state: ResearchAgentState) -> str:
    """根据当前阶段路由到下一步或结束"""
    
    # 如果有错误，保持在当前阶段
    if state.get("error"):
        return state["current_phase"]
    
    # 根据当前阶段决定下一步
    current = state["current_phase"]
    
    if current == "perception":
        return "modeling"
    elif current == "modeling":
        return "reasoning"
    elif current == "reasoning":
        return "decision"
    elif current == "decision":
        return "report"
    elif current == "report":
        return END
    else:
        return END

# 创建智能体工作流图
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()

# 测试函数
def run_research_agent(topic: str, industry: str, horizon: str) -> Dict[str, Any]:
    """运行研究智能体并返回结果"""
    
    # 创建工作流
    agent = create_research_agent_workflow()
    
    # 准备初始状态
    initial_state = {
        "research_topic": topic,
        "industry_focus": industry,
        "time_horizon": horizon,
        "perception_data": None,
        "world_model": None,
        "reasoning_plans": None,
        "selected_plan": None,
        "final_report": None,
        "current_phase": "perception",
        "error": None
    }
    print("LangGraph Mermaid流程图：")
    print(agent.get_graph().draw_mermaid())
    
    # 运行智能体
    result = agent.invoke(initial_state)
    
    return result

# 主函数
if __name__ == "__main__":
    print("=== 深思熟虑智能体 - 智能投研助手 ===\n")
    print("使用模型：Qwen-Turbo-2025-04-28\n")
    
    # 用户输入
    topic = input("请输入研究主题 (例如: 新能源汽车行业投资机会): ")
    industry = input("请输入行业焦点 (例如: 电动汽车制造、电池技术): ")
    horizon = input("请输入时间范围 [短期/中期/长期]: ")
    
    print("\n智能投研助手开始工作...\n")
    
    try:
        # 运行智能体
        result = run_research_agent(topic, industry, horizon)
        
        # 处理结果
        if result.get("error"):
            print(f"\n发生错误: {result['error']}")
        else:
            print("\n=== 最终研究报告 ===\n")
            print(result.get("final_report", "未生成报告"))
            
            # 保存报告
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"research_report_{timestamp}.txt"
            
            with open(filename, "w", encoding="utf-8") as f:
                f.write(result.get("final_report", "未生成报告"))
            
            print(f"\n报告已保存为: {filename}")
            
    except Exception as e:
        print(f"\n运行过程中发生错误: {str(e)}")