Skip to content

AI Agent在金融分析中的实践

概述

AI Agent（智能代理）是在 LLM 基础上构建的自主决策系统，能够感知环境、制定计划、调用工具并迭代执行任务。在金融分析领域，AI Agent 可以自主完成数据获取、指标计算、报告生成等复杂工作流 (Workflow)，从"问答式助手"进化为"自主分析员"。本文探讨金融 AI Agent 的架构设计、工具调用、多智能体系统 (Multi-Agent Systems) 和工作流编排 (Workflow Orchestration)。

自主分析 Agent (Autonomous Analysis Agent)

Agent 的核心循环

Agent 的运行遵循 Observe-Think-Act 循环：

\[\text{Agent}: s_{t+1} = \text{LLM}(\text{Observation}_t, \text{Memory}_t, \text{Tools})\]
┌──────────────┐
│  用户指令     │
└──────┬───────┘
       ▼
┌──────────────┐     ┌──────────────┐
│  思考/规划    │────▶│  工具调用     │
│  (Reasoning) │     │  (Tool Use)  │
└──────┬───────┘     └──────┬───────┘
       ▲                    │
       │    ┌───────────┐   │
       └────│  观察结果  │◀──┘
            │ (Observe) │
            └───────────┘

基础架构实现

class FinancialAnalystAgent:
    def __init__(self, llm_client, tools, memory_size=20):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
        self.memory = []
        self.memory_size = memory_size

    def run(self, task):
        """执行金融分析任务"""
        self.memory.append({"role": "user", "content": task})
        max_iterations = 10

        for i in range(max_iterations):
            # Think: LLM 决定下一步行动
            response = self.llm.generate(
                messages=self.memory,
                tools=self.get_tool_schemas(),
                system_prompt=FINANCIAL_AGENT_SYSTEM_PROMPT
            )

            # Act: 如果需要调用工具
            if response.has_tool_calls:
                for tool_call in response.tool_calls:
                    tool_name = tool_call.name
                    tool_args = tool_call.arguments
                    result = self.tools[tool_name].execute(**tool_args)
                    self.memory.append({
                        "role": "tool",
                        "name": tool_name,
                        "content": str(result)
                    })
            else:
                # 任务完成，返回最终回答
                return response.content

            # 控制记忆长度
            if len(self.memory) > self.memory_size:
                self.memory = self.memory[:2] + self.memory[-self.memory_size+2:]

        return "达到最大迭代次数，任务未完成。"

System Prompt 的设计

金融 Agent 的 System Prompt 需要明确：(1) 角色定位（如"资深金融分析师"）；(2) 可用工具及其使用时机；(3) 分析框架和方法论；(4) 安全约束（如不提供投资建议）。

工具调用 (Tool Use for Data Retrieval)

金融数据工具集

class StockDataTool:
    name = "get_stock_data"
    description = "获取股票的历史行情数据和基本面指标"

    def execute(self, ticker, start_date, end_date, fields=None):
        import yfinance as yf
        stock = yf.Ticker(ticker)
        hist = stock.history(start=start_date, end=end_date)
        info = stock.info
        return {
            "price_data": hist.to_dict(),
            "fundamentals": {
                "pe_ratio": info.get('trailingPE'),
                "market_cap": info.get('marketCap'),
                "revenue": info.get('totalRevenue'),
                "profit_margin": info.get('profitMargins')
            }
        }

class FinancialCalculatorTool:
    name = "financial_calculator"
    description = "执行金融计算: DCF估值、比率分析、收益率计算等"

    def execute(self, calculation_type, **params):
        if calculation_type == "dcf":
            return self._dcf_valuation(**params)
        elif calculation_type == "ratio_analysis":
            return self._ratio_analysis(**params)
        elif calculation_type == "returns":
            return self._calculate_returns(**params)

    def _dcf_valuation(self, fcf, growth_rate, wacc, terminal_growth, years=5):
        pv_fcf = sum(
            fcf * (1 + growth_rate)**t / (1 + wacc)**t
            for t in range(1, years + 1)
        )
        terminal_fcf = fcf * (1 + growth_rate)**years * (1 + terminal_growth)
        terminal_value = terminal_fcf / (wacc - terminal_growth)
        pv_terminal = terminal_value / (1 + wacc)**years
        return {"enterprise_value": pv_fcf + pv_terminal}

class NewsSearchTool:
    name = "search_news"
    description = "搜索特定公司或主题的最新金融新闻"

    def execute(self, query, date_range="7d", max_results=10):
        # 调用新闻API
        results = news_api.search(query, date_range, max_results)
        return [{"title": r.title, "summary": r.summary,
                 "date": r.date, "source": r.source} for r in results]

class ReportGeneratorTool:
    name = "generate_report"
    description = "生成格式化的金融分析报告"

    def execute(self, title, sections, format="markdown"):
        report = f"# {title}\n\n"
        for section in sections:
            report += f"## {section['heading']}\n\n{section['content']}\n\n"
        return report

工具安全与权限控制

金融 Agent 的工具调用需严格限制：(1) 只读权限——Agent 不应能执行交易；(2) 数据访问边界——限制可查询的数据范围；(3) 速率限制——防止过度调用外部 API；(4) 审计日志——记录所有工具调用以便追溯。

多智能体系统 (Multi-Agent Systems)

分工协作架构

class MultiAgentFinancialTeam:
    def __init__(self):
        self.coordinator = CoordinatorAgent()
        self.agents = {
            "macro_analyst": MacroAnalystAgent(),
            "equity_analyst": EquityAnalystAgent(),
            "quant_analyst": QuantAnalystAgent(),
            "risk_analyst": RiskAnalystAgent(),
        }

    def analyze(self, task):
        """多智能体协作分析"""
        # Step 1: 协调器分解任务
        subtasks = self.coordinator.decompose(task)

        # Step 2: 分配给专业 Agent
        results = {}
        for subtask in subtasks:
            agent_name = subtask['assigned_to']
            agent = self.agents[agent_name]
            results[agent_name] = agent.run(subtask['description'])

        # Step 3: 协调器综合结果
        final_report = self.coordinator.synthesize(results)
        return final_report

Agent 角色定义

Agent 职责 工具集
宏观分析师 宏观经济分析、政策解读 经济数据API、政策文件检索
行业分析师 行业趋势、竞争格局 行业数据库、研报检索
量化分析师 因子分析、回测 Python执行、数据库查询
风险分析师 风险评估、压力测试 风险模型、情景模拟
协调器 任务分解、结果综合 内部通信

Agent 间通信

class AgentMessage:
    def __init__(self, sender, receiver, content, msg_type="info"):
        self.sender = sender
        self.receiver = receiver
        self.content = content
        self.msg_type = msg_type  # info / request / response / alert

class MessageBus:
    def __init__(self):
        self.messages = []

    def send(self, message):
        self.messages.append(message)
        # 触发接收方处理
        return self.route(message)

    def route(self, message):
        if message.msg_type == "request":
            # 请求类消息: 路由到目标Agent
            return target_agent.handle_request(message)
        elif message.msg_type == "alert":
            # 警报类消息: 广播给所有Agent
            return [agent.handle_alert(message) for agent in self.agents]

工作流编排 (Workflow Orchestration)

定义分析流水线

class FinancialAnalysisWorkflow:
    def __init__(self):
        self.steps = [
            {"name": "data_collection", "agent": "quant_analyst",
             "task": "收集目标公司的财务数据和行情数据"},
            {"name": "fundamental_analysis", "agent": "equity_analyst",
             "task": "基于收集的数据进行基本面分析"},
            {"name": "macro_context", "agent": "macro_analyst",
             "task": "分析当前宏观环境对目标公司的影响"},
            {"name": "risk_assessment", "agent": "risk_analyst",
             "task": "评估投资的主要风险因素",
             "depends_on": ["fundamental_analysis", "macro_context"]},
            {"name": "synthesis", "agent": "coordinator",
             "task": "综合所有分析结果，生成投资研究报告",
             "depends_on": ["risk_assessment"]}
        ]

    def execute(self, target_company):
        results = {}
        for step in self.topological_sort(self.steps):
            # 注入前序步骤的结果作为上下文
            context = {dep: results[dep]
                       for dep in step.get('depends_on', [])}
            agent = self.agents[step['agent']]
            results[step['name']] = agent.run(
                step['task'],
                context=context,
                target=target_company
            )
        return results['synthesis']

工作流的容错设计

金融分析工作流需要考虑：(1) 某个 Agent 失败时的重试和回退机制；(2) 中间结果的缓存以避免重复计算；(3) 人工审核节点 (Human-in-the-Loop) 在关键决策点的介入；(4) 超时控制以防止 Agent 陷入死循环。

评估与监控

\[\text{Agent Quality} = w_1 \cdot \text{Accuracy} + w_2 \cdot \text{Efficiency} + w_3 \cdot \text{Safety}\]

关键监控指标：

  • 任务完成率：Agent 成功完成指定任务的比例
  • 工具调用效率：达成目标所需的平均工具调用次数
  • 幻觉率：输出中包含事实错误的比例
  • 延迟：从任务提交到完成的时间

小结

AI Agent 将 LLM 从被动的问答工具升级为主动的分析系统。在金融领域，Agent 的核心价值在于自动化复杂的分析工作流——从数据获取到报告生成。多智能体架构通过专业分工提升了分析的深度和广度。然而，Agent 的自主性也带来了安全和可控性挑战，严格的权限管理、审计追溯和人工监督机制是金融 Agent 系统不可或缺的组成部分。