Single Agent 和 Multi-Agent 究竟怎么选，用科研助手 + 基金分析两个实战项目带你分析

# tools_fund.py
from langchain.tools import tool
import json
from datetime import date


@tool
def fund_basic_info(code: str) -> str:
    """获取基金的基础信息：成立时间、规模、费率、类型。

    输入：基金代码字符串，如 "000001"。
    输出：JSON 字符串，包含字段: code, name, type, inception_date, size, fee。
    """
    mock = {
        "code": code,
        "name": "示例成长混合",
        "type": "混合型",
        "inception_date": "2015-05-20",
        "size": "35.2亿",
        "fee": "1.5%",
    }
    return json.dumps(mock, ensure_ascii=False)


@tool
def fund_nav_history(code: str) -> str:
    """获取基金历史净值（简化版）。

    输入：基金代码。
    输出：JSON 列表，每个元素包含 date, nav。
    """
    mock = [
        {"date": "2024-11-20", "nav": 1.23},
        {"date": "2024-11-21", "nav": 1.25},
        {"date": "2024-11-22", "nav": 1.22},
    ]
    return json.dumps(mock, ensure_ascii=False)


@tool
def index_history(symbol: str) -> str:
    """获取指数历史走势（简化版）。"""
    mock = [
        {"date": "2024-11-20", "close": 4200},
        {"date": "2024-11-21", "close": 4220},
        {"date": "2024-11-22", "close": 4150},
    ]
    return json.dumps(mock, ensure_ascii=False)


@tool
def news_search(keyword: str) -> str:
    """按关键词搜索最近的市场新闻（简化版）。"""
    mock_news = [
        {"title": "宏观经济数据略弱，市场短期承压", "date": str(date.today())},
        {"title": f"与 {keyword} 相关的行业出现资金流入迹象", "date": str(date.today())},
    ]
    return json.dumps(mock_news, ensure_ascii=False)

# single_agent_fund.py
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from tools_fund import fund_basic_info, fund_nav_history, index_history, news_search

SYSTEM_PROMPT = """
你是一个基金分析助手，**不构成任何投资建议**。

用户会输入类似：
- "帮我看看 000001 这只基金"
- "这只医药主题 ETF 还能不能定投"

你的工作流程：
1. 如果用户没有说明风险承受能力，你要先询问（保守 / 中性 / 激进）。
2. 调用工具获取：
   - fund_basic_info: 基金基本面（成立时间、规模、费率、类型）
   - fund_nav_history: 最近一段时间净值走势
   - index_history: 相关指数走势（如果能从基金类型推断）
   - news_search: 相关行业的最新新闻
3. 分析：
   - 基金本身的波动特征（波动大 / 小）
   - 与大盘的联动情况
   - 费率是否偏高
   - 对不同风险偏好用户的大致适配性
4. 输出一份结构化报告，包括：
   - 基金概览
   - 历史表现简评
   - 风险提示（重点）
   - 对不同类型投资者的中性描述（不要给出 "买入/卖出" 结论）

严禁：
- 做出具体买卖建议；
- 保证收益。
"""

def build_fund_single_agent():
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
    tools = [fund_basic_info, fund_nav_history, index_history, news_search]

    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        handle_parsing_errors=True,
        verbose=True,
        agent_kwargs={"system_message": SYSTEM_PROMPT},
    )
    return agent


if __name__ == "__main__":
    agent = build_fund_single_agent()
    res = agent.invoke("帮我分析一下 000001 这只基金，看看适不适合作为长期定投标的？我风险偏好偏中性。")
    print(res["output"])

# graph_fund.py
from typing import TypedDict, Dict, Any, List
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from tools_fund import fund_basic_info, fund_nav_history, index_history, news_search
import json


class FundState(TypedDict):
    code: str
    risk_profile: str  # "conservative" / "neutral" / "aggressive"
    basic_info: Dict[str, Any]
    nav_history: List[Dict[str, Any]]
    index_history: List[Dict[str, Any]]
    news: List[Dict[str, Any]]
    analysis: str
    advice: str


def data_agent_node(state: FundState) -> FundState:
    """DataAgent：只负责拉取各种数据。"""
    code = state["code"]
    b = json.loads(fund_basic_info.invoke(code))
    nav = json.loads(fund_nav_history.invoke(code))
    idx = json.loads(index_history.invoke("SH300"))
    news = json.loads(news_search.invoke(code))

    new_state = dict(state)
    new_state["basic_info"] = b
    new_state["nav_history"] = nav
    new_state["index_history"] = idx
    new_state["news"] = news
    return new_state


def analysis_agent_node(state: FundState) -> FundState:
    """AnalysisAgent：根据数据做波动/回撤/相关性等分析。"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
    prompt = f"""
你是一个量化分析助手。

基金基础信息：
{json.dumps(state['basic_info'], ensure_ascii=False, indent=2)}

基金净值历史：
{json.dumps(state['nav_history'], ensure_ascii=False, indent=2)}

相关指数历史：
{json.dumps(state['index_history'], ensure_ascii=False, indent=2)}

相关市场新闻：
{json.dumps(state['news'], ensure_ascii=False, indent=2)}

请输出一段「纯分析文本」，内容包括：
- 基金波动特征（大/小）
- 与指数走势的关系
- 历史回撤特点（定性即可）
- 该基金更偏向什么风格（稳健/成长/主题投机等）

只做分析，不要给投资建议。
"""
    resp = llm.invoke(prompt)
    new_state = dict(state)
    new_state["analysis"] = resp.content
    return new_state


def advice_agent_node(state: FundState) -> FundState:
    """AdviceAgent：根据分析结果 + 风险偏好，生成「非投资建议」报告。"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    prompt = f"""
你是一个合规的基金分析助手，**不能给出任何具体买卖建议**。

用户风险偏好：{state['risk_profile']}

量化分析结论如下：
{state['analysis']}

请生成一份面向普通投资者的中文报告，包含：
1. 基金概览（用通俗语言复述一下）
2. 历史表现与波动特征
3. 对不同风险偏好投资者可能的适配性分析：
   - 保守型投资者：
   - 中性型投资者：
   - 激进型投资者：
4. 风险提示（务必写清楚）

禁止出现：
- “建议买入/卖出/加仓/减仓”等措辞；
- 保证收益的表述。
"""
    resp = llm.invoke(prompt)
    new_state = dict(state)
    new_state["advice"] = resp.content
    return new_state


def build_fund_graph():
    g = StateGraph(FundState)
    g.add_node("data_agent", data_agent_node)
    g.add_node("analysis_agent", analysis_agent_node)
    g.add_node("advice_agent", advice_agent_node)

    g.add_edge(START, "data_agent")
    g.add_edge("data_agent", "analysis_agent")
    g.add_edge("analysis_agent", "advice_agent")
    g.add_edge("advice_agent", END)

    return g.compile()


if __name__ == "__main__":
    graph = build_fund_graph()
    init_state: FundState = {
        "code": "000001",
        "risk_profile": "neutral",
        "basic_info": {},
        "nav_history": [],
        "index_history": [],
        "news": [],
        "analysis": "",
        "advice": "",
    }

    final = graph.invoke(init_state)
    print("=== 基金分析报告 ===")
    print(final["advice"])