LangGraph 高级教程：多 Agent 股票分析系统

基于 LangGraph 构建的多 Agent 协作系统，解决股票投资价值分析问题

🎯 项目简介

本项目演示如何使用 LangGraph 构建多 Agent 协作系统。

解决的任务：

帮我分析一下茅台股票的投资价值，包括行业前景、财务数据、风险评估，最后生成一份投资报告。

解决方案：四个专业 Agent 协作完成

用户问题 → 行业研究员 → 财务分析师 → 风险评估师 → 报告撰写者
              ↓              ↓              ↓              ↓
           行业前景       财务数据       风险评估       最终报告

Agent	职责	工具
行业研究员	调研行业前景、市场规模、竞争格局	get_industry_info
财务分析师	分析财务指标、估值水平	get_financial_data, get_stock_price
风险评估师	评估估值风险、政策风险	get_risk_factors
报告撰写者	汇总生成结构化投资报告	LLM

---

🚀 快速开始（新手指南）

第一步：环境准备

确保你的电脑已安装：

• Python 3.10+（检查方式：python3 --version）
• Git（检查方式：git --version）

第二步：克隆项目

git clone https://github.com/helloworldtang/langgraph-tutorials.git
cd langgraph-tutorials

第三步：安装 uv

uv 是一个快速的 Python 包管理工具（类似 pip，但更快）。

macOS / Linux：

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows（PowerShell）：

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

安装后重启终端，验证：

uv --version

第四步：创建虚拟环境并安装依赖

# 创建虚拟环境
uv venv

# 激活虚拟环境
source .venv/bin/activate    # macOS / Linux
# 或
.venv\Scripts\activate       # Windows

# 安装依赖
uv add langchain langchain-openai langchain-core langgraph python-dotenv pydantic pytest

第五步：配置 API Key

1. 复制环境变量模板：

cp .env.example .env

2. 获取 DeepSeek API Key：

   • 访问 https://platform.deepseek.com/
   • 注册/登录后，在「API Keys」页面创建新密钥

3. 编辑 .env 文件，填入你的 API Key：

DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxx

第六步：运行演示

uv run python demos/main.py

预期输出：

============================================================
🎯 任务目标：分析股票投资价值
============================================================
📌 目标股票：茅台（600519）
============================================================

============================================================
🔄 [行业研究员] 开始工作
   正在查询行业前景信息...
============================================================
...

============================================================
📝 最终投资报告
============================================================

### 贵州茅台（600519）投资报告

**1. 投资摘要**
结论：维持"买入"评级...

第七步：运行测试

uv run pytest tests/ -v

预期输出：

============================= test session starts ==============================
...
tests/test_langgraph.py::TestState::test_state_keys PASSED
...
============================== 12 passed ==============================

---

📁 项目结构

langgraph-tutorials/
├── src/
│   ├── __init__.py
│   ├── state/                   # 状态管理
│   │   └── __init__.py           # StockAnalysisState 定义
│   ├── agents/                   # Agent 节点实现
│   │   ├── __init__.py           # 四个专业 Agent 节点 + 工作流
│   │   └── tools.py              # 工具定义（行业、财务、风险）
│   └── workflow/                 # 工作流封装
│       └── __init__.py           # StockAnalysisWorkflow
├── demos/
│   └── main.py                   # 演示入口
├── tests/
│   └── test_langgraph.py         # 单元测试
├── .env.example                  # 环境变量模板
├── pyproject.toml                # 项目配置
└── README.md

---

🔧 核心代码说明

1. 状态定义（State）

所有 Agent 共享同一个状态，数据自动累积：

# src/state/__init__.py

class StockAnalysisState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add]  # 消息历史，自动累积
    next_agent: str                                  # 下一个要执行的节点
    industry_research: str                          # 行业前景
    financial_data: str                             # 财务数据
    risk_assessment: str                            # 风险评估
    final_report: str                               # 最终报告

2. 节点定义（Node）

每个节点是独立的函数，输入状态，输出更新后的状态：

# src/agents/__init__.py

def industry_researcher_node(state: StockAnalysisState) -> StockAnalysisState:
    """行业前景研究员"""
    # 1. 调用工具获取行业信息
    result = get_industry_info.invoke({"industry": "白酒"})
    
    # 2. 返回更新后的状态
    return {
        "messages": [AIMessage(content=result)],
        "next_agent": "financial_analyst",  # 下一个节点
        "industry_research": result
    }

3. 工作流构建（Graph）

# 使用 LangGraph 构建工作流
workflow = StateGraph(StockAnalysisState)

# 添加节点
workflow.add_node("industry_researcher", industry_researcher_node)
workflow.add_node("financial_analyst", financial_analyst_node)
workflow.add_node("risk_assessor", risk_assessor_node)
workflow.add_node("report_writer", report_writer_node)

# 定义边（流程）
workflow.set_entry_point("industry_researcher")
workflow.add_edge("industry_researcher", "financial_analyst")
workflow.add_edge("financial_analyst", "risk_assessor")
workflow.add_edge("risk_assessor", "report_writer")
workflow.add_edge("report_writer", END)

# 编译并运行
app = workflow.compile()
result = app.invoke(initial_state)

---

🆚 LangGraph vs 函数式编程

方面	函数式编程	LangGraph 工作流
流程控制	固定顺序	可动态路由
状态管理	手动传参	自动共享
条件跳转	if-else	conditional_edges
循环支持	手动循环	原生支持
调试	逐步追踪	节点级调试
适用场景	简单任务	多 Agent 协作

何时用 LangGraph：

• ✅ 多 Agent 协作
• ✅ 需要条件路由（根据结果决定下一步）
• ✅ 需要循环（结果不满足可以重新执行）
• ✅ 复杂状态管理

何时用函数式：

• ✅ 流程固定
• ✅ 简单任务，步骤少

---

📈 进阶方向

方向	具体建议
工具扩展	接入真实金融 API（Wind、Choice），获取实时行情
Agent 增强	增加宏观研究员、竞争对手对比分析师
循环支持	让风险评估师可以要求行业研究员重新补充信息
记忆持久化	使用 Redis 存储历史分析报告
流式输出	使用 stream() 逐步展示每个节点的结果
多股票支持	抽象为通用框架，支持任意股票分析

---

❓ 常见问题

Q1: uv: command not found

安装 uv 后需要重启终端，或执行：

source ~/.bashrc    # 或 ~/.zshrc

Q2: DEEPSEEK_API_KEY not found

确保 .env 文件在项目根目录，且已填入有效的 API Key。

Q3: 测试失败

确保已激活虚拟环境并安装所有依赖：

source .venv/bin/activate
uv add pytest

---

🛠 环境要求

• Python 3.10+
• DeepSeek API Key（获取地址）

---

📝 License

MIT