第 1 节：完整项目代码与运行

1. 下载和启动

下载完整项目 ZIP

pip install -r requirements.txt
python main.py

在解压后的项目目录执行。输入：小米集团最近有什么值得关注的信息？。默认走本地模拟工具，因此无需 API Key、行情 API 或联网搜索服务。

2. 要用真实模型时，直接改代码顶部

打开 main.py，填写服务商发放的密钥，设置模型名；把 MOCK_MODE 改为 False。兼容 OpenAI 的服务按服务商说明填写 BASE_URL。

MOCK_MODE = False
OPENAI_API_KEY = "在这里填写你自己的 API Key"
OPENAI_MODEL = "gpt-4o-mini"
OPENAI_BASE_URL = ""

不要把填写真实密钥的文件提交到公共 Git 仓库。没有密钥就保持模拟模式。

3. 完整代码：main.py

以下代码和下载 ZIP 中的 main.py 一致，复制时保留完整内容。依赖版本在 ZIP 的 requirements.txt。

"""ApacheCN 教学示例：LangChain + LangGraph 投研工作流。
示例工具只返回虚构模拟数据，不提供投资建议。
"""
import os
import sys
from typing import Literal, TypedDict

from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field

# ===== 在这里配置 =====
MOCK_MODE = True  # 不配置密钥先保持 True；要调用模型时改为 False
OPENAI_API_KEY = ""  # 填入服务商发放的 API Key；不要提交含真实密钥的代码
OPENAI_MODEL = "gpt-4o-mini"
OPENAI_BASE_URL = ""  # OpenAI 兼容服务可填写服务商提供的地址；官方接口留空
# =====================

if OPENAI_API_KEY:
    os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY


class ResearchQuery(BaseModel):
    """从用户问题抽取研究计划，不确定的证券代码留空。"""

    company: str | None = Field(default=None, description="公司或证券名称")
    market: Literal["CN", "HK", "US", "unknown"] = "unknown"
    symbol: str | None = Field(default=None, description="证券代码；无法确认时为 null")
    needs_quote: bool = Field(default=False, description="是否询问行情")
    needs_search: bool = Field(default=False, description="是否需要最新资料")


class AgentState(TypedDict, total=False):
    question: str
    query: dict
    quote: dict | None
    sources: list[dict]
    answer: str
    errors: list[str]
    cache_hit: bool


@tool
def mock_quote(company: str) -> dict:
    """获取本教程内置的模拟行情，绝非实时数据。"""
    return {
        "label": "模拟行情",
        "company": company,
        "price": 68.50,
        "change_percent": 1.25,
        "as_of": "虚构示例日期",
        "source": "本地示例数据",
    }


@tool
def mock_search(company: str) -> list[dict]:
    """查询本教程内置的示例资料，不会访问互联网。"""
    if "无结果" in company:
        return []
    return [{
        "label": "模拟资料",
        "title": f"{company}：用于演示资料引用的虚构条目",
        "url": "https://example.invalid/mock-source",
        "summary": "这是程序内置的示例，不是新闻或真实公司信息。",
    }]


def make_model():
    if MOCK_MODE:
        return None
    if not OPENAI_API_KEY:
        raise ValueError("请在文件顶部填写 OPENAI_API_KEY，或将 MOCK_MODE 设为 True。")
    kwargs = {"temperature": 0}
    if OPENAI_BASE_URL:
        kwargs["base_url"] = OPENAI_BASE_URL
    return init_chat_model(f"openai:{OPENAI_MODEL}", **kwargs)


QUERY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "只提取用户明确提到的公司信息，不猜证券代码。判断是否需要行情或最新资料。"),
    ("human", "{question}"),
])
ANSWER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "用中文回答。仅依据给定资料陈述事实；资料不足就明确说明。模拟数据必须标注为模拟，不提供投资建议。"),
    ("human", "问题：{question}\n解析：{query}\n行情：{quote}\n资料：{sources}"),
])


def parse_question(state: AgentState) -> dict:
    question = state["question"]
    if MOCK_MODE:
        company = next((name for name in ("小米集团", "腾讯", "阿里巴巴", "比亚迪") if name in question), None)
        needs_quote = any(word in question for word in ("股价", "行情", "价格"))
        needs_search = any(word in question for word in ("最近", "新闻", "信息", "关注"))
        market = "HK" if company in ("小米集团", "腾讯", "阿里巴巴") else "CN" if company else "unknown"
        parsed = ResearchQuery(company=company, market=market, needs_quote=needs_quote, needs_search=needs_search)
    else:
        model = make_model()
        parsed = (QUERY_PROMPT | model.with_structured_output(ResearchQuery)).invoke({"question": question})
    return {"query": parsed.model_dump()}


def route(state: AgentState) -> str:
    if state.get("cache_hit"):
        return "answer"
    query = state.get("query", {})
    if not query.get("company"):
        return "answer"
    if query.get("needs_quote"):
        return "quote"
    if query.get("needs_search"):
        return "search"
    return "answer"


def check_cache(state: AgentState) -> dict:
    cached = ANSWER_CACHE.get(state["question"])
    return {"cache_hit": cached is not None, "answer": cached}


def after_quote(state: AgentState) -> str:
    return "search" if state.get("query", {}).get("needs_search") else "answer"


def get_quote(state: AgentState) -> dict:
    try:
        return {"quote": mock_quote.invoke({"company": state["query"]["company"]})}
    except Exception as exc:
        return {"quote": None, "errors": [f"行情工具异常：{exc}"]}


def search(state: AgentState) -> dict:
    try:
        rows = mock_search.invoke({"company": f"{state['query']['company']} {state['question']}"})
        return {"sources": rows}
    except Exception as exc:
        return {"sources": [], "errors": [f"搜索工具异常：{exc}"]}


def answer(state: AgentState) -> dict:
    if state.get("cache_hit") and state.get("answer"):
        return {"answer": state["answer"]}
    query = state.get("query", {})
    quote = state.get("quote")
    sources = state.get("sources", [])
    if not query.get("company"):
        text = "我还不能确定你问的是哪家公司，请补充公司全名或证券名称。"
    elif not quote and not sources:
        text = "没有取得可核实的资料，因此无法作事实判断。请补充信息或稍后重试。"
    elif MOCK_MODE:
        text = (
            f"识别结果：{query['company']}（市场：{query['market']}）。\n"
            f"行情：{quote or '本次未查询行情'}\n资料：{sources or '本次未查询资料'}\n"
            "以上全部是本地虚构模拟数据，不是实时行情或新闻，也不构成投资建议。"
        )
    else:
        try:
            text = (ANSWER_PROMPT | make_model()).invoke({
                "question": state["question"], "query": query,
                "quote": quote, "sources": sources,
            }).content
        except Exception as exc:
            text = f"模型调用失败，当前无法形成回答：{exc}"
    ANSWER_CACHE[state["question"]] = text
    return {"answer": text}


def build_graph():
    graph = StateGraph(AgentState)
    graph.add_node("parse", parse_question)
    graph.add_node("cache", check_cache)
    graph.add_node("quote", get_quote)
    graph.add_node("search", search)
    graph.add_node("answer", answer)
    graph.add_edge(START, "parse")
    graph.add_edge("parse", "cache")
    graph.add_conditional_edges("cache", route, {
        "quote": "quote", "search": "search", "answer": "answer",
    })
    graph.add_conditional_edges("quote", after_quote, {"search": "search", "answer": "answer"})
    graph.add_edge("search", "answer")
    graph.add_edge("answer", END)
    return graph.compile()


ANSWER_CACHE: dict[str, str] = {}


def main():
    question = " ".join(sys.argv[1:]).strip() or input("你想了解哪家公司？ ").strip()
    if not question:
        print("请输入一个问题。")
        return
    try:
        result = build_graph().invoke({"question": question, "sources": [], "errors": []})
        print("\n" + result.get("answer", "没有生成回答。"))
        for error in result.get("errors", []):
            print("处理记录：" + error)
    except Exception as exc:
        print(f"运行失败：{exc}")


if __name__ == "__main__":
    main()

4. 预期结果和边界

程序输出识别公司、市场和模拟资料，并醒目标出这些是虚构示例。输入“请查无结果测试公司的新闻”会得到资料不足说明；不能识别公司时会请用户补充名称。

这里没有接真实行情或搜索服务，示例数值不能用于任何投资判断。