LangGraph实战：构建具备长期记忆与复杂推理的金融问答Agent

在实际 AI 应用开发中，构建一个能够理解复杂指令、利用工具、并保持长期记忆的智能体（Agent）是核心挑战。开发者常常面临工具链集成困难、状态管理复杂、调试过程不透明以及生产部署繁琐等问题。LangChain 及其生态，包括 LangGraph，正是为了解决这些工程化难题而生的开源框架和平台。它们提供了一套标准化的组件和开发范式，让开发者能够更专注于业务逻辑，而非底层基础设施的搭建。本文将深入探讨 LangChain 和 LangGraph 的核心概念、区别，并通过一个完整的“金融大模型问答机器人”项目案例，展示如何从零开始，使用 Qwen 大模型、LangChain、LangGraph、RAG 等技术栈，构建一个具备知识检索、长期记忆和复杂推理能力的生产级 AI 应用。无论你是希望快速入门 LangChain，还是想了解如何将 LangGraph 用于构建可靠的多步骤 Agent，本文都将提供从环境准备、核心代码实现到生产部署考量的全链路实践指南。

1. 理解 LangChain 与 LangGraph：从快速构建到可靠编排

在开始项目之前，必须厘清 LangChain 和 LangGraph 的定位与关系。它们是互补而非替代的关系，服务于 Agent 开发的不同阶段和需求层次。

1.1 LangChain：AI 应用开发的“脚手架”与“粘合剂”

LangChain 的核心目标是降低 AI 应用，特别是基于大语言模型（LLM）的应用的开发门槛。它通过提供一系列标准化的“链”（Chains）、“工具”（Tools）和“记忆”（Memory）等抽象，将 LLM 与外部数据源、计算逻辑和状态管理连接起来。

通俗地讲，LangChain 就像一套高度模块化的乐高积木。它预先定义好了各种连接器（如调用 OpenAI API、读取 PDF、查询数据库的接口）和标准件（如对话记忆模块、文本分割器）。开发者可以像搭积木一样，快速组合出一个能完成特定任务的 AI 应用原型，例如一个简单的文档问答机器人。它的优势在于“快速启动”和“丰富的生态集成”。

在技术定义上，LangChain 主要包含以下核心概念：

• 模型 I/O（Model I/O）： 统一不同 LLM 供应商（如 OpenAI、Anthropic、本地模型）的调用接口。
• 检索（Retrieval）： 构建和管理外部知识库（如向量数据库），实现检索增强生成（RAG）。
• 链（Chains）： 将多个 LLM 调用或其他工具调用按顺序组合起来，完成更复杂的任务。
• 代理（Agents）： 让 LLM 根据用户目标，自主决定调用哪些工具以及调用的顺序，这是 LangChain 实现“智能”的关键。
• 记忆（Memory）： 在对话或多次调用间持久化状态信息，如聊天历史。

一个典型的 LangChain 快速入门示例是构建一个简单的问答链：

# 示例：使用 LangChain 快速构建一个问答链 from langchain_openai import ChatOpenAI from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate # 1. 初始化模型 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 2. 定义提示词模板 prompt = ChatPromptTemplate.from_template("请用中文回答：{question}") # 3. 创建链 chain = LLMChain(llm=llm, prompt=prompt) # 4. 运行链 response = chain.invoke({"question": "LangChain 是什么？"}) print(response["text"])

这个例子展示了 LangChain 的“粘合”作用：它将模型、提示词和调用逻辑封装成一个可执行的chain对象。然而，当任务流程变得复杂、需要循环、条件分支或精确的状态控制时，仅靠基础的Chain和Agent会显得力不从心，代码可读性和可维护性下降。这时就需要 LangGraph。

1.2 LangGraph：基于状态图的可靠 Agent 编排引擎

LangGraph 是建立在 LangChain 之上的一个库，它引入了有状态、可循环的计算图这一核心抽象。你可以把它想象成专门为 AI Agent 设计的“工作流引擎”或“状态机框架”。

它的设计动机是解决复杂、多步骤 Agent 的可靠性问题。传统的 LangChain Agent 执行路径是隐式的，由 LLM 每次决定下一步，调试困难且难以保证确定性。LangGraph 则要求开发者显式地定义整个 Agent 的工作流（图），节点代表操作（调用 LLM、工具），边代表状态流转的条件。

技术定义上，LangGraph 的核心是StateGraph。开发者需要：

1. 定义一个State类型，明确描述 Agent 在每个步骤所拥有的全部信息（如用户输入、已调用工具的结果、聊天历史等）。
2. 创建多个Node（节点），每个节点是一个函数，接收State，修改State并返回更新后的State。
3. 定义Edge（边），决定在某个节点执行完毕后，下一步应该走到哪个节点。边可以是固定的，也可以根据State的内容动态决定（条件边）。

这种范式带来了几个关键优势：

• 确定性： 工作流是预先定义好的图，执行路径清晰可预测。
• 可调试性： 每个节点的输入输出（State）都是明确的，便于追踪和日志记录。
• 复杂逻辑： 原生支持循环（比如让 Agent 反复思考直到满意）、并行、条件分支等复杂控制流。
• 持久化与容错： 由于整个 Agent 的状态被封装在State对象中，因此可以很方便地将状态序列化保存到数据库（如 SQLite），实现长期记忆和故障恢复。

1.3 LangChain 与 LangGraph 的核心区别与选型建议

为了更清晰地对比，我们将两者的核心差异总结如下表：

选型建议：

• 如果你的需求是：快速验证一个想法，构建一个简单的文档问答、文本总结或一次性的数据处理流程，从 LangChain 开始。它的高级Agent和Chain已经足够强大。
• 如果你的需求是：构建一个需要反复与用户或工具交互、有严格步骤顺序（如先检索、再分析、最后生成报告）、或者需要将对话状态保存数月之久的客服机器人或虚拟助手，那么应该直接使用 LangGraph。它为生产环境的可靠性提供了必要的基础设施。

我们的“金融大模型问答机器人”项目，因为涉及知识检索、多轮对话记忆和可能的多步骤推理（例如，先查行情，再分析风险，最后给出建议），正是一个适合使用LangGraph来构建的典型案例。接下来，我们将进入实战环节。

2. 项目实战：构建金融大模型问答机器人

本项目将模拟一个为内部员工或客户服务的金融问答助手。它能回答关于金融产品、市场术语、公司政策等知识库内的问题，并能记住对话历史，进行连贯的多轮交流。当问题超出知识库范围时，它能礼貌地拒绝或引导提问。

2.1 项目设计与技术栈选型

项目目标：开发一个可通过 API 调用的智能金融问答服务，具备准确的知识检索、流畅的多轮对话和清晰的拒绝回答能力。

核心能力设计：

1. 知识检索增强（RAG）： 将金融知识文档（PDF、Word等）向量化存储，提问时优先从知识库中检索相关片段作为上下文。
2. 长期记忆： 将每段对话的历史记录持久化到 SQLite 数据库，实现跨会话的记忆。
3. 智能路由与编排： 使用 LangGraph 构建一个工作流，根据用户问题决定是进行知识库问答、闲聊还是拒绝回答。
4. 大模型集成： 使用通义千问（Qwen）作为核心 LLM，兼顾效果与成本。

技术栈详解：

• LLM：Qwen。选择其 API 版本（如qwen-max）或本地部署版本。相比 OpenAI API，它更适合中文场景且成本可控。本文示例将使用Qwen2.5-7B-Instruct的本地化调用方式。
• 应用框架：FastAPI。提供高性能、异步的 RESTful API 接口，方便前端或其他服务集成。
• AI 编排框架：LangChain + LangGraph。LangChain 提供基础的模型调用、提示词管理、文本分割和向量检索组件；LangGraph 用于构建可靠的问答工作流。
• 向量数据库/检索：Chroma（本地轻量级）或PGVector（生产级）。本文为简化演示，使用LangChain内置的Chroma和OpenAIEmbeddings（需替换为兼容 Qwen 的嵌入模型，如text2vec）。
• 长期记忆存储：SQLite。利用 LangChain 的SQLChatMessageHistory将对话历史存入 SQLite 数据库。
• 其他关键技术：
  • RAG（Retrieval-Augmented Generation）： 项目核心模式。
  • GraphRAG： 一个更高级的 RAG 概念，强调利用图结构来组织和检索知识。本项目暂不深入，但知识库可以视为其基础。
  • 高效微调/LoRA/SFT： 用于后续垂直领域效果优化。本文聚焦于应用开发，微调部分仅作方向性说明。
  • PPO/DPO/知识蒸馏/量化： 属于模型优化和部署阶段的进阶技术，不在本次基础构建范围内。

2.2 环境准备与依赖配置

首先，创建一个干净的 Python 虚拟环境并安装核心依赖。

# 创建并激活虚拟环境（以 conda 为例） conda create -n finance_agent python=3.10 conda activate finance_agent # 安装核心依赖 pip install langchain langchain-community langgraph pip install fastapi uvicorn sqlalchemy pip install chromadb pypdf sentence-transformers # 用于向量化和文档加载 pip install “pydantic>=2.0” # LangChain 对 Pydantic 版本有要求 # 安装大模型相关依赖（以使用 Ollama 本地运行 Qwen 为例） # 首先确保已安装并运行 Ollama，并在其中拉取 Qwen 模型: `ollama pull qwen2.5:7b` pip install ollama langchain-ollama

关键依赖版本说明：

• langchain和langgraph： 确保使用较新版本（如>=0.2.0），其 API 相对稳定。
• sentence-transformers： 用于生成文本向量。我们使用text2vec模型，它对中文友好且无需 API 密钥。
• ollama： 一个本地运行大模型的工具，方便快速测试。生产环境可考虑直接调用 Qwen API 或部署独立的模型服务。

项目目录结构：

finance_qa_agent/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── graph/ │ │ ├── __init__.py │ │ ├── state.py # 定义 Graph 的状态 │ │ ├── nodes.py # 定义各个节点函数 │ │ └── graph.py # 构建并编译 LangGraph │ ├── memory/ │ │ ├── __init__.py │ │ └── sqlite_chat_history.py # 封装 SQLite 记忆存储 │ ├── retriever/ │ │ ├── __init__.py │ │ └── vector_store.py # 向量库初始化与检索逻辑 │ └── config.py # 配置文件 ├── data/ # 存放知识库文档 │ └── financial_knowledge.pdf ├── storage/ # 存储向量数据库和 SQLite 文件 │ ├── chroma_db/ │ └── chat_history.db ├── requirements.txt └── README.md

2.3 核心模块实现：记忆、检索与图状态

在构建工作流之前，我们需要先实现几个基础设施模块。

1. 记忆模块 (app/memory/sqlite_chat_history.py)： 此模块负责将对话历史持久化到 SQLite，实现长期记忆。

from langchain_community.chat_message_histories import SQLChatMessageHistory from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine import os class ChatHistoryManager: """管理对话历史的类，每个会话一个独立的 history 对象""" def __init__(self, db_path: str = “./storage/chat_history.db”): # 确保存储目录存在 os.makedirs(os.path.dirname(db_path), exist_ok=True) # SQLite 连接字符串 self.connection_string = f“sqlite:///{db_path}” # 创建引擎（echo=True 可查看 SQL 日志，调试用） self.engine = create_engine(self.connection_string, echo=False) def get_history_for_session(self, session_id: str) -> SQLChatMessageHistory: """根据会话ID获取或创建对应的聊天历史记录""" # SQLChatMessageHistory 会自动创建表 return SQLChatMessageHistory( session_id=session_id, connection_string=self.connection_string ) def clear_history(self, session_id: str): """清除指定会话的历史记录（可选功能）""" history = self.get_history_for_session(session_id) history.clear() # 全局记忆管理器实例 memory_manager = ChatHistoryManager()

2. 检索模块 (app/retriever/vector_store.py)： 此模块负责加载知识文档、创建向量库，并提供检索功能。

from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma import os class VectorStoreRetriever: def __init__(self, persist_directory: str = “./storage/chroma_db”, embedding_model_name: str = “shibing624/text2vec-base-chinese”): self.persist_directory = persist_directory self.embedding_model = HuggingFaceEmbeddings( model_name=embedding_model_name, model_kwargs={‘device’: ‘cpu’}, # 根据环境改为 ‘cuda’ encode_kwargs={‘normalize_embeddings’: True} ) self.vector_store = None self._init_vector_store() def _init_vector_store(self): """初始化或加载已有的向量库""" if os.path.exists(self.persist_directory) and os.listdir(self.persist_directory): # 加载已存在的向量库 self.vector_store = Chroma( persist_directory=self.persist_directory, embedding_function=self.embedding_model ) print(f“向量库已从 {self.persist_directory} 加载。”) else: # 创建新的空向量库 self.vector_store = Chroma( persist_directory=self.persist_directory, embedding_function=self.embedding_model ) print(f“新的向量库已在 {self.persist_directory} 创建。”) def add_documents(self, file_path: str): """向向量库添加文档（如PDF）""" loader = PyPDFLoader(file_path) documents = loader.load() # 分割文本 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=[“\n\n”, “\n”, “。”, “；”, “，”, “ ”, “”] ) splits = text_splitter.split_documents(documents) # 添加到向量库 self.vector_store.add_documents(splits) # 持久化 self.vector_store.persist() print(f“已成功添加文档 {file_path}，共 {len(splits)} 个文本块。”) def retrieve(self, query: str, k: int = 3) -> list: """检索与查询最相关的 k 个文档片段""" if self.vector_store is None: return [] docs = self.vector_store.similarity_search(query, k=k) return [doc.page_content for doc in docs] # 全局检索器实例 retriever = VectorStoreRetriever() # 首次运行时可添加知识文档 # retriever.add_documents(“./data/financial_knowledge.pdf”)

3. 图状态定义 (app/graph/state.py)： 这是 LangGraph 的核心，它定义了工作流中流转的“状态”对象的结构。

from typing import TypedDict, List, Optional, Annotated import operator class GraphState(TypedDict): """ 定义 LangGraph 工作流的状态。 所有节点都读取和修改这个状态字典。 """ # 用户输入的问题 question: str # 从向量库检索到的上下文 context: Optional[List[str]] # 当前的对话历史（从记忆模块加载） chat_history: List[str] # 工作流最终生成的答案 answer: Optional[str] # 一个标志位，用于控制流程走向（例如，是否需要检索） needs_retrieval: bool

这里我们使用了TypedDict来定义状态的结构。Annotated和operator的导入是为后续可能的状态合并操作做准备。needs_retrieval是一个关键字段，它将用于决定工作流的走向。

2.4 构建 LangGraph 工作流：节点与编排

接下来，我们将定义工作流的各个节点，并将它们组装成一个完整的图。

1. 定义节点函数 (app/graph/nodes.py)： 每个节点都是一个纯函数，接收GraphState，返回更新后的GraphState。

from app.graph.state import GraphState from app.retriever.vector_store import retriever from app.memory.sqlite_chat_history import memory_manager from langchain_ollama import ChatOllama from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # 初始化本地 Qwen 模型（通过 Ollama） llm = ChatOllama(model=“qwen2.5:7b”, temperature=0.2) def retrieve_node(state: GraphState) -> GraphState: """检索节点：从向量库获取相关上下文""" print(f“检索节点：正在检索问题 ‘{state[‘question’]}’ 的相关信息...”) context = retriever.retrieve(state[‘question’]) return {**state, “context”: context} def decide_route_node(state: GraphState) -> GraphState: """路由决策节点：根据问题和上下文，决定是否需要检索或直接生成""" question = state[‘question’] chat_history = state[‘chat_history’] # 这里可以设计更复杂的路由逻辑，例如： # 1. 如果是问候语（如“你好”），不需要检索。 # 2. 如果问题非常泛泛（如“介绍一下金融”），可能需要检索。 # 3. 如果上下文为空且问题专业，需要检索。 # 本例使用一个简单规则：如果问题长度小于5个字符或包含特定问候词，则不检索。 simple_greetings = [“你好”, “嗨”, “hello”, “hi”, “在吗”] needs_retrieval = True if len(question.strip()) < 5 or any(greeting in question for greeting in simple_greetings): needs_retrieval = False print(f“路由决策节点：问题 ‘{question}’， 决定是否需要检索 -> {needs_retrieval}”) return {**state, “needs_retrieval”: needs_retrieval} def generate_answer_node(state: GraphState) -> GraphState: """生成答案节点：结合历史、上下文和问题，调用 LLM 生成最终答案""" question = state[‘question’] context = state.get(‘context’, []) chat_history = state[‘chat_history’] # 构建提示词 system_prompt = “““你是一个专业的金融问答助手。请根据用户的问题和提供的参考信息（如果有）进行回答。 回答要求： 1. 使用中文，简洁、专业、准确。 2. 如果参考信息与问题相关，请基于参考信息回答。 3. 如果参考信息不相关或为空，请根据你的知识回答，并说明这是通用知识。 4. 如果问题完全超出你的能力或知识范围，请礼貌地表示无法回答，并建议用户咨询相关专业人士。 5. 请考虑对话历史，使回答连贯。 ”“” # 格式化对话历史 formatted_history = “\n”.join(chat_history[-6:]) if chat_history else “无” # 格式化检索到的上下文 formatted_context = “\n\n”.join(context) if context else “未提供相关参考信息。” prompt = ChatPromptTemplate.from_messages([ (“system”, system_prompt), MessagesPlaceholder(variable_name=“history”), (“human”, “对话历史：\n{formatted_history}\n\n参考信息：\n{formatted_context}\n\n用户当前问题：{question}”) ]) # 准备消息列表 messages = [ SystemMessage(content=system_prompt), ] # 添加历史消息（这里简化处理，实际应转换为 Message 对象） for msg in chat_history[-4:]: # 只取最近几轮历史 # 简单假设历史记录是交替的 Human/AI 消息字符串 # 实际项目中应存储结构化消息 pass messages.append(HumanMessage(content=f“历史：{formatted_history}\n上下文：{formatted_context}\n问题：{question}”)) # 调用模型 print(“生成答案节点：正在调用 LLM 生成答案...”) response = llm.invoke(messages) answer = response.content # 更新状态 new_history = chat_history + [f“用户：{question}”, f“助手：{answer}”] return {**state, “answer”: answer, “chat_history”: new_history} def save_memory_node(state: GraphState) -> GraphState: """保存记忆节点：将本轮对话存入 SQLite 数据库""" # 注意：在实际的 LangGraph 运行中，我们通常在一个统一的入口管理记忆。 # 这里为了演示，假设我们有一个全局的 session_id。 # 更佳实践是在 State 中传入 session_id，并在此节点调用 memory_manager。 session_id = “default_session” # 应从外部传入（如 API 请求头） history_obj = memory_manager.get_history_for_session(session_id) # 将本轮对话添加到历史（这里简化，实际应添加 HumanMessage 和 AIMessage 对象） # history_obj.add_user_message(state[‘question’]) # history_obj.add_ai_message(state[‘answer’]) print(“保存记忆节点：对话历史已保存（模拟）。”) return state

2. 构建并编译图 (app/graph/graph.py)： 这是 LangGraph 的核心，我们将节点连接起来，定义工作流。

from langgraph.graph import StateGraph, END from app.graph.state import GraphState from app.graph.nodes import retrieve_node, decide_route_node, generate_answer_node, save_memory_node def create_finance_agent_graph(): """创建并编译金融问答 Agent 的工作流图""" # 1. 初始化一个状态图，指定状态类型 workflow = StateGraph(GraphState) # 2. 添加节点 workflow.add_node(“decide_route”, decide_route_node) # 决策节点 workflow.add_node(“retrieve”, retrieve_node) # 检索节点 workflow.add_node(“generate”, generate_answer_node) # 生成节点 workflow.add_node(“save_memory”, save_memory_node) # 记忆节点（可选） # 3. 设置入口点 workflow.set_entry_point(“decide_route”) # 4. 添加边，定义流程逻辑 # 从 decide_route 出发，根据 needs_retrieval 的值决定下一步 workflow.add_conditional_edges( “decide_route”, # 这是一个路由函数，根据 state 返回下一个节点的名称 lambda state: “retrieve” if state.get(“needs_retrieval”, True) else “generate”, { “retrieve”: “retrieve”, # 如果返回 “retrieve”，则跳转到 retrieve 节点 “generate”: “generate” # 如果返回 “generate”，则跳转到 generate 节点 } ) # retrieve 节点之后，总是进入 generate 节点 workflow.add_edge(“retrieve”, “generate”) # generate 节点之后，可以进入 save_memory 节点，然后结束 workflow.add_edge(“generate”, “save_memory”) workflow.add_edge(“save_memory”, END) # 也可以直接从 generate 节点结束（如果不需立即保存） # workflow.add_edge(“generate”, END) # 5. 编译图 app = workflow.compile() return app # 创建全局的图应用实例 finance_agent_app = create_finance_agent_graph()

这个图定义了一个清晰的工作流：

1. 从decide_route开始，判断问题是否需要检索知识库。
2. 如果需要 (needs_retrieval=True)，则执行retrieve节点获取上下文。
3. 无论是否检索，最终都会进入generate节点，结合所有信息生成答案。
4. 生成答案后，可选择进入save_memory节点持久化历史，然后流程结束。

2.5 集成 FastAPI 并提供服务

最后，我们使用 FastAPI 将 LangGraph 工作流包装成一个 HTTP API 服务。

FastAPI 主应用 (app/main.py)：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.graph.graph import finance_agent_app from app.memory.sqlite_chat_history import memory_manager from typing import List, Optional app = FastAPI(title=“金融问答机器人 API”, description=“基于 LangGraph 和 Qwen 的智能金融问答服务”) class ChatRequest(BaseModel): """聊天请求体""" question: str session_id: str = “default_session” # 用于区分不同用户的对话 use_history: bool = True # 是否使用历史记录 class ChatResponse(BaseModel): """聊天响应体""" answer: str session_id: str context_used: Optional[List[str]] = None # 返回使用的上下文，便于调试 @app.post(“/chat”, response_model=ChatResponse) async def chat_endpoint(request: ChatRequest): """处理用户提问的核心端点""" try: # 1. 从记忆库加载该会话的历史记录 chat_history = [] if request.use_history: history_obj = memory_manager.get_history_for_session(request.session_id) # 这里简化处理，实际应从 history_obj.messages 中提取字符串 # 假设我们有一个方法 to_list() 返回字符串列表 # chat_history = history_obj.to_list() pass # 2. 准备 LangGraph 的初始状态 initial_state = { “question”: request.question, “context”: None, “chat_history”: chat_history, “answer”: None, “needs_retrieval”: True # 初始值，会被 decide_route 节点覆盖 } # 3. 执行工作流图 print(f“开始处理会话 ‘{request.session_id}’ 的请求: {request.question}”) final_state = finance_agent_app.invoke(initial_state) # 4. 保存本轮对话到记忆库（实际应在 save_memory 节点完成，这里做备份） if request.use_history: history_obj = memory_manager.get_history_for_session(request.session_id) # history_obj.add_user_message(request.question) # history_obj.add_ai_message(final_state[“answer”]) # 5. 返回响应 return ChatResponse( answer=final_state[“answer”], session_id=request.session_id, context_used=final_state.get(“context”) ) except Exception as e: print(f“处理请求时发生错误: {e}”) raise HTTPException(status_code=500, detail=f“服务器内部错误: {str(e)}”) @app.get(“/health”) async def health_check(): """健康检查端点""" return {“status”: “healthy”} if __name__ == “__main__”: import uvicorn uvicorn.run(app, host=“0.0.0.0”, port=8000)

2.6 运行验证与测试

现在，我们可以启动服务并进行测试。

1. 启动服务：

   cd finance_qa_agent python -m app.main

   服务将在http://0.0.0.0:8000启动。

2. 测试 API： 使用curl或 Postman 等工具发送请求。

   curl -X POST “http://localhost:8000/chat" \ -H “Content-Type: application/json” \ -d ‘{“question”: “什么是市盈率？”, “session_id”: “user_123”}’

   预期返回一个 JSON 响应，包含answer字段。

3. 查看文档： 访问http://localhost:8000/docs可以查看自动生成的交互式 API 文档（Swagger UI），方便直接测试。

3. 关键配置、参数详解与生产考量

3.1 核心参数调优

在开发和生产中，以下参数需要根据实际情况调整：

3.2 生产环境部署建议

学习环境可以快速跑通，但生产环境需要考虑更多：

1. 向量数据库升级： 将 Chroma 替换为PGVector（PostgreSQL 扩展）或Milvus/Weaviate等专业向量数据库，以获得更好的性能、可扩展性和高可用性。
2. 大模型服务化： 使用vLLM、TGI或OpenAI API 兼容的接口来部署 Qwen 模型，提供稳定、高性能的模型推理服务，而非在应用进程中直接调用 Ollama。
3. API 服务增强：
   • 认证与鉴权： 使用 JWT、OAuth2 等机制保护 API。
   • 限流与熔断： 使用 FastAPI 中间件或 API 网关（如 Kong, Nginx）防止滥用。
   • 异步处理： 对于耗时的请求，可引入消息队列（如 Celery + Redis）进行异步处理，并返回任务 ID 供客户端轮询。
4. 可观测性： 集成LangSmith。这是 LangChain 官方提供的平台，可以无缝追踪 LangGraph 工作流的每一步执行情况，记录输入输出，进行效果评估，极大提升调试和迭代效率。
5. 配置外置： 将模型路径、API 密钥、数据库连接等配置信息移至环境变量或配置中心（如 Apollo, Nacos）。

4. 常见问题排查与优化方向

4.1 常见问题排查清单

在开发和运行过程中，你可能会遇到以下问题：

4.2 性能与效果优化方向

1. 检索优化（RAG 核心）：
   • 混合检索： 结合向量检索（语义相似）和关键词检索（BM25），提高召回率。
   • 重排序（Re-ranking）： 使用交叉编码器模型对检索出的文档进行精排，将最相关的放在前面。
   • 元数据过滤： 在向量检索时加入文档来源、章节等元数据过滤条件。
2. 提示词工程：
   • 少样本（Few-Shot）提示： 在系统提示词中提供几个高质量的问答示例，引导模型输出格式和风格。
   • 思维链（Chain-of-Thought）： 对于复杂推理问题，提示模型先一步步思考，再给出最终答案。
3. 引入 LangSmith：
   • 这是提升开发效率的“神器”。通过 LangSmith，你可以可视化整个工作流的执行轨迹，查看每个节点的输入输出，对不同的提示词或检索策略进行效果评估和对比，快速定位问题所在。
4. 模型微调：
   • 如果通用模型在金融领域的术语、逻辑或格式上表现不佳，可以考虑使用LoRA等技术对 Qwen 模型进行高效微调，使其更贴合专业领域。
5. 图工作流复杂化：
   • 当前是简单的“检索-生成”两段式。可以引入更多节点，例如：
     • 问题分类节点： 更精细地路由到不同处理子图（如查询股价、解释术语、生成报告）。
     • 查询改写节点： 根据对话历史改写当前问题，使其更适合检索。
     • 答案验证节点： 调用另一个 LLM 或规则引擎对生成的答案进行事实核查。

通过以上步骤，你不仅构建了一个可运行的金融问答机器人，更掌握了一套基于 LangChain 和 LangGraph 构建生产级 AI Agent 的方法论。从明确 LangGraph 的状态图思想，到实现记忆、检索等核心模块，再到通过 FastAPI 提供服务，最后考虑生产部署和优化，这条路径适用于大多数需要复杂编排和状态管理的智能体场景。记住，可靠的 AI 应用不仅仅是模型调用，更是对数据流、状态和业务逻辑的精心编排。