使用AI智能体构建金融分析代理 2026

简介

能够分析金融数据并生成投资报告的AI智能体正在改变开发者的工作方式。本指南将教你如何使用LangGraph + 计算器工具构建一个生产就绪的金融分析代理。

我们要构建什么

一个能够完成以下任务的金融分析代理：

理解复杂请求

将请求分解为子任务

自主执行任务

处理错误并重试

生成一致、高质量的输出

架构

用户请求
    ↓
[金融分析代理编排器]
    ↓
[任务规划] → [工具选择] → [执行]
    ↓                                      ↓
[验证] ←──────────────────── [结果]
    ↓
最终输出

使用LangGraph实现

python
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
import json
状态定义
class FinancialAnalysisAgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    task: str
    sub_tasks: List[str]
    completed_tasks: List[str]
    final_output: str | None
    iterations: int
定义用于分析金融数据和生成投资报告的工具
@tool
def analyze_task(task: str) -> str:
    """将复杂任务分解为子任务。"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(f"将此任务分解为3-5个具体、可操作的子任务：{task}")
    return response.content
@tool
def execute_sub_task(sub_task: str, context: str = "") -> str:
    """执行特定的子任务。"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(
        f"上下文：{context}\n\n执行此特定任务：{sub_task}\n提供详细输出。"
    )
    return response.content
@tool
def validate_output(task: str, output: str) -> str:
    """验证输出是否满足要求。"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(
        f"任务：{task}\n\n待验证的输出：{output}\n\n"
        f"此输出是否完整且正确？如果不完整，缺少什么？"
    )
    return response.content
tools = [analyze_task, execute_sub_task, validate_output]
初始化带工具的LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
llm_with_tools = llm.bind_tools(tools)
智能体节点
def agent_node(state: FinancialAnalysisAgentState):
    if state.get("iterations", 0) > 8:
        return {"final_output": "达到最大迭代次数", "iterations": 9}
    
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iterations": state.get("iterations", 0) + 1
    }
工具执行节点
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
def should_continue(state: FinancialAnalysisAgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
        return "tools"
    return "end"
构建图
workflow = StateGraph(FinancialAnalysisAgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
workflow.add_edge("tools", "agent")
agent = workflow.compile()

使用智能体

python
from langchain_core.messages import HumanMessage
def run_financial_analysis_agent(request: str) -> str:
    """在用户请求上运行金融分析代理。"""
    
    initial_state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent.invoke(initial_state)
    
    # 提取最终答案
    last_message = result["messages"][-1]
    return last_message.content
使用示例
output = run_financial_analysis_agent(
    "创建一个全面的计划，用于分析金融数据并生成投资报告"
)
print(output)

添加持久化记忆

python
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("./agent_memory.db") as checkpointer:
    agent_with_memory = workflow.compile(checkpointer=checkpointer)
def run_with_memory(request: str, session_id: str) -> str:
    config = {"configurable": {"thread_id": session_id}}
    
    state = {
        "messages": [HumanMessage(content=request)],
        "task": request,
        "sub_tasks": [],
        "completed_tasks": [],
        "final_output": None,
        "iterations": 0
    }
    
    result = agent_with_memory.invoke(state, config=config)
    return result["messages"][-1].content
第一次交互
response1 = run_with_memory("开始分析金融数据并生成投资报告", session_id="session-001")
后续交互（智能体记住上下文）
response2 = run_with_memory("从上次中断的地方继续", session_id="session-001")

生产环境：FastAPI服务

python
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
app = FastAPI(title="金融分析代理服务")
class AgentRequest(BaseModel):
    task: str
    session_id: str = "default"
    stream: bool = False
@app.post("/agent/run")
async def run_agent(request: AgentRequest):
    if request.stream:
        async def stream_response():
            async for event in agent.astream_events(
                {"messages": [HumanMessage(content=request.task)], 
                 "task": request.task, "sub_tasks": [], 
                 "completed_tasks": [], "final_output": None, "iterations": 0},
                version="v2"
            ):
                if event["event"] == "on_chat_model_stream":
                    content = event["data"]["chunk"].content
                    if content:
                        yield content
        
        return StreamingResponse(stream_response(), media_type="text/plain")
    
    result = run_financial_analysis_agent(request.task)
    return {"result": result, "session_id": request.session_id}
@app.get("/health")
async def health():
    return {"status": "healthy", "agent": "金融分析代理"}

监控智能体性能

python
from dataclasses import dataclass
from datetime import datetime
import statistics
@dataclass
class AgentMetrics:
    task: str
    iterations: int
    duration_ms: float
    success: bool
    output_length: int
metrics_store: List[AgentMetrics] = []
def run_with_metrics(task: str) -> tuple[str, AgentMetrics]:
    import time
    start = time.time()
    
    try:
        result = run_financial_analysis_agent(task)
        success = True
    except Exception as e:
        result = f"错误：{e}"
        success = False
    
    duration = (time.time() - start) * 1000
    
    # 注意：在生产环境中，迭代次数来自实际状态
    metrics = AgentMetrics(
        task=task[:50],
        iterations=3,  
        duration_ms=duration,
        success=success,
        output_length=len(result)
    )
    metrics_store.append(metrics)
    return result, metrics
def print_metrics_report():
    if not metrics_store:
        return
    
    successful = [m for m in metrics_store if m.success]
    durations = [m.duration_ms for m in metrics_store]
    
    print(f"总运行次数：{len(metrics_store)}")
    print(f"成功率：{len(successful)/len(metrics_store):.1%}")
    print(f"平均耗时：{statistics.mean(durations):.0f}ms")
    print(f"p95耗时：{sorted(durations)[int(len(durations)*0.95)]:.0f}ms")

最佳实践

限制迭代次数：始终设置最大值以防止无限循环

检查点状态：对长时间运行的任务使用持久化

人工审核：对关键操作添加审批步骤

详细日志：记录每次工具调用以便调试

优雅失败：处理错误而不导致智能体崩溃

结论

使用AI智能体构建金融分析代理能够自主分析金融数据并生成投资报告。LangGraph实现为生产使用提供了控制与灵活性的恰当平衡。

从一个简单的概念验证开始，添加持久化，然后随着信心的增长逐步扩展。

---

*使用LangGraph + 计算器工具实现的金融分析代理 | 2026年5月*