在实际 AI 应用开发中构建一个能够理解复杂指令、利用工具、并保持长期记忆的智能体Agent是核心挑战。开发者常常面临工具链集成困难、状态管理复杂、调试过程不透明以及生产部署繁琐等问题。LangChain 及其生态包括 LangGraph正是为了解决这些工程化难题而生的开源框架和平台。它们提供了一套标准化的组件和开发范式让开发者能够更专注于业务逻辑而非底层基础设施的搭建。本文将深入探讨 LangChain 和 LangGraph 的核心概念、区别并通过一个完整的“金融大模型问答机器人”项目案例展示如何从零开始使用 Qwen 大模型、LangChain、LangGraph、RAG 等技术栈构建一个具备知识检索、长期记忆和复杂推理能力的生产级 AI 应用。无论你是希望快速入门 LangChain还是想了解如何将 LangGraph 用于构建可靠的多步骤 Agent本文都将提供从环境准备、核心代码实现到生产部署考量的全链路实践指南。1. 理解 LangChain 与 LangGraph从快速构建到可靠编排在开始项目之前必须厘清 LangChain 和 LangGraph 的定位与关系。它们是互补而非替代的关系服务于 Agent 开发的不同阶段和需求层次。1.1 LangChainAI 应用开发的“脚手架”与“粘合剂”LangChain 的核心目标是降低 AI 应用特别是基于大语言模型LLM的应用的开发门槛。它通过提供一系列标准化的“链”Chains、“工具”Tools和“记忆”Memory等抽象将 LLM 与外部数据源、计算逻辑和状态管理连接起来。通俗地讲LangChain 就像一套高度模块化的乐高积木。它预先定义好了各种连接器如调用 OpenAI API、读取 PDF、查询数据库的接口和标准件如对话记忆模块、文本分割器。开发者可以像搭积木一样快速组合出一个能完成特定任务的 AI 应用原型例如一个简单的文档问答机器人。它的优势在于“快速启动”和“丰富的生态集成”。在技术定义上LangChain 主要包含以下核心概念模型 I/OModel 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(modelgpt-3.5-turbo, temperature0) # 2. 定义提示词模板 prompt ChatPromptTemplate.from_template(请用中文回答{question}) # 3. 创建链 chain LLMChain(llmllm, promptprompt) # 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。开发者需要定义一个State类型明确描述 Agent 在每个步骤所拥有的全部信息如用户输入、已调用工具的结果、聊天历史等。创建多个Node节点每个节点是一个函数接收State修改State并返回更新后的State。定义Edge边决定在某个节点执行完毕后下一步应该走到哪个节点。边可以是固定的也可以根据State的内容动态决定条件边。这种范式带来了几个关键优势确定性 工作流是预先定义好的图执行路径清晰可预测。可调试性 每个节点的输入输出State都是明确的便于追踪和日志记录。复杂逻辑 原生支持循环比如让 Agent 反复思考直到满意、并行、条件分支等复杂控制流。持久化与容错 由于整个 Agent 的状态被封装在State对象中因此可以很方便地将状态序列化保存到数据库如 SQLite实现长期记忆和故障恢复。1.3 LangChain 与 LangGraph 的核心区别与选型建议为了更清晰地对比我们将两者的核心差异总结如下表特性维度LangChainLangGraph核心抽象链Chain、代理Agent、工具Tool状态图State Graph、节点Node、边Edge控制流相对线性或由 LLM 隐式决策复杂控制流实现较繁琐显式定义原生支持循环、条件分支、并行等复杂控制流状态管理通过Memory类管理通常与对话历史相关通过强类型的State对象集中管理所有上下文信息调试难度复杂 Agent 的中间步骤和决策原因较难追踪执行路径清晰每个节点的输入输出状态可见易于调试适用场景快速原型、简单问答、标准 RAG、基础工具调用复杂多步骤任务、需精确编排的 Agent、需持久化状态的长期对话、生产级可靠系统学习曲线相对平缓入门简单需要理解图计算和状态机概念曲线更陡峭与 LangChain 关系基础框架基于 LangChain 构建的扩展库用于高级编排选型建议如果你的需求是快速验证一个想法构建一个简单的文档问答、文本总结或一次性的数据处理流程从 LangChain 开始。它的高级Agent和Chain已经足够强大。如果你的需求是构建一个需要反复与用户或工具交互、有严格步骤顺序如先检索、再分析、最后生成报告、或者需要将对话状态保存数月之久的客服机器人或虚拟助手那么应该直接使用 LangGraph。它为生产环境的可靠性提供了必要的基础设施。我们的“金融大模型问答机器人”项目因为涉及知识检索、多轮对话记忆和可能的多步骤推理例如先查行情再分析风险最后给出建议正是一个适合使用LangGraph来构建的典型案例。接下来我们将进入实战环节。2. 项目实战构建金融大模型问答机器人本项目将模拟一个为内部员工或客户服务的金融问答助手。它能回答关于金融产品、市场术语、公司政策等知识库内的问题并能记住对话历史进行连贯的多轮交流。当问题超出知识库范围时它能礼貌地拒绝或引导提问。2.1 项目设计与技术栈选型项目目标开发一个可通过 API 调用的智能金融问答服务具备准确的知识检索、流畅的多轮对话和清晰的拒绝回答能力。核心能力设计知识检索增强RAG 将金融知识文档PDF、Word等向量化存储提问时优先从知识库中检索相关片段作为上下文。长期记忆 将每段对话的历史记录持久化到 SQLite 数据库实现跨会话的记忆。智能路由与编排 使用 LangGraph 构建一个工作流根据用户问题决定是进行知识库问答、闲聊还是拒绝回答。大模型集成 使用通义千问Qwen作为核心 LLM兼顾效果与成本。技术栈详解LLMQwen。选择其 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 数据库。其他关键技术RAGRetrieval-Augmented Generation 项目核心模式。GraphRAG 一个更高级的 RAG 概念强调利用图结构来组织和检索知识。本项目暂不深入但知识库可以视为其基础。高效微调/LoRA/SFT 用于后续垂直领域效果优化。本文聚焦于应用开发微调部分仅作方向性说明。PPO/DPO/知识蒸馏/量化 属于模型优化和部署阶段的进阶技术不在本次基础构建范围内。2.2 环境准备与依赖配置首先创建一个干净的 Python 虚拟环境并安装核心依赖。# 创建并激活虚拟环境以 conda 为例 conda create -n finance_agent python3.10 conda activate finance_agent # 安装核心依赖 pip install langchain langchain-community langgraph pip install fastapi uvicorn sqlalchemy pip install chromadb pypdf sentence-transformers # 用于向量化和文档加载 pip install “pydantic2.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.md2.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_okTrue) # SQLite 连接字符串 self.connection_string f“sqlite:///{db_path}” # 创建引擎echoTrue 可查看 SQL 日志调试用 self.engine create_engine(self.connection_string, echoFalse) def get_history_for_session(self, session_id: str) - SQLChatMessageHistory: 根据会话ID获取或创建对应的聊天历史记录 # SQLChatMessageHistory 会自动创建表 return SQLChatMessageHistory( session_idsession_id, connection_stringself.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_nameembedding_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_directoryself.persist_directory, embedding_functionself.embedding_model ) print(f“向量库已从 {self.persist_directory} 加载。”) else: # 创建新的空向量库 self.vector_store Chroma( persist_directoryself.persist_directory, embedding_functionself.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_size500, chunk_overlap50, 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, kk) 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”, temperature0.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(contentsystem_prompt), ] # 添加历史消息这里简化处理实际应转换为 Message 对象 for msg in chat_history[-4:]: # 只取最近几轮历史 # 简单假设历史记录是交替的 Human/AI 消息字符串 # 实际项目中应存储结构化消息 pass messages.append(HumanMessage(contentf“历史{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 state2. 构建并编译图 (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()这个图定义了一个清晰的工作流从decide_route开始判断问题是否需要检索知识库。如果需要 (needs_retrievalTrue)则执行retrieve节点获取上下文。无论是否检索最终都会进入generate节点结合所有信息生成答案。生成答案后可选择进入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_modelChatResponse) 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( answerfinal_state[“answer”], session_idrequest.session_id, context_usedfinal_state.get(“context”) ) except Exception as e: print(f“处理请求时发生错误: {e}”) raise HTTPException(status_code500, detailf“服务器内部错误: {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”, port8000)2.6 运行验证与测试现在我们可以启动服务并进行测试。启动服务cd finance_qa_agent python -m app.main服务将在http://0.0.0.0:8000启动。测试 API 使用curl或 Postman 等工具发送请求。curl -X POST “http://localhost:8000/chat \ -H “Content-Type: application/json” \ -d ‘{“question”: “什么是市盈率”, “session_id”: “user_123”}’预期返回一个 JSON 响应包含answer字段。查看文档 访问http://localhost:8000/docs可以查看自动生成的交互式 API 文档Swagger UI方便直接测试。3. 关键配置、参数详解与生产考量3.1 核心参数调优在开发和生产中以下参数需要根据实际情况调整模块参数说明建议值/调整方向文本分割chunk_size文本块大小字符数。太小丢失上下文太大检索不准。金融文档建议 300-800可测试不同大小对检索效果的影响。chunk_overlap块间重叠字符数。保证上下文连贯。一般为chunk_size的 10%-20%。向量检索k(top-k)每次检索返回的文档片段数量。从 3 开始尝试根据答案质量和速度平衡。复杂问题可增至 5-7。嵌入模型文本转向量的模型。中文场景首选text2vec系列。生产环境考虑BGE等更优模型。大模型temperature生成答案的随机性。0 最确定1 最随机。金融问答建议较低值如 0.1-0.3保证答案稳定性。max_tokens生成答案的最大长度。根据问题复杂度设置如 512 或 1024。图工作流路由逻辑decide_route_node中的判断逻辑。可根据问题分类模型、关键词匹配或意图识别来优化提高路由准确率。记忆历史长度传入 LLM 的对话历史轮数。太多会消耗 Token 且可能干扰一般取最近 3-6 轮。3.2 生产环境部署建议学习环境可以快速跑通但生产环境需要考虑更多向量数据库升级 将 Chroma 替换为PGVectorPostgreSQL 扩展或Milvus/Weaviate等专业向量数据库以获得更好的性能、可扩展性和高可用性。大模型服务化 使用vLLM、TGI或OpenAI API 兼容的接口来部署 Qwen 模型提供稳定、高性能的模型推理服务而非在应用进程中直接调用 Ollama。API 服务增强认证与鉴权 使用 JWT、OAuth2 等机制保护 API。限流与熔断 使用 FastAPI 中间件或 API 网关如 Kong, Nginx防止滥用。异步处理 对于耗时的请求可引入消息队列如 Celery Redis进行异步处理并返回任务 ID 供客户端轮询。可观测性 集成LangSmith。这是 LangChain 官方提供的平台可以无缝追踪 LangGraph 工作流的每一步执行情况记录输入输出进行效果评估极大提升调试和迭代效率。配置外置 将模型路径、API 密钥、数据库连接等配置信息移至环境变量或配置中心如 Apollo, Nacos。4. 常见问题排查与优化方向4.1 常见问题排查清单在开发和运行过程中你可能会遇到以下问题问题现象可能原因检查方式处理建议启动服务时报错ModuleNotFoundError依赖未安装或虚拟环境未激活。检查requirements.txt和当前 Python 环境。在正确的虚拟环境中运行pip install -r requirements.txt。调用/chatAPI 返回 500 错误日志显示连接失败。Ollama 服务未启动或模型未下载。1. 运行ollama list检查模型是否存在。2. 运行ollama serve启动服务。3. 检查main.py中模型名称是否正确。确保 Ollama 服务在运行且已通过ollama pull qwen2.5:7b下载模型。问答响应慢。1. 本地模型首次加载慢。2. 向量检索耗时。3. 网络问题。1. 观察日志看时间消耗在哪个环节。2. 检查向量库文档数量是否过多。1. 预热模型。2. 为向量检索建立索引。3. 考虑使用 GPU 运行嵌入模型和 LLM。答案与知识库内容无关。1. 检索到的上下文不相关。2. 提示词未正确引导模型使用上下文。1. 检查retrieve函数返回的context内容。2. 在提示词中明确要求“基于以下参考信息回答”。1. 调整文本分割参数或尝试不同的嵌入模型。2. 优化提示词使用更严格的格式如“参考信息[context]\n问题[question]”。多轮对话中模型“忘记”了之前的内容。1. 记忆未正确保存或加载。2. 传入 LLM 的历史消息格式错误或长度被截断。1. 检查 SQLite 数据库中对应session_id是否有记录。2. 打印generate_answer_node中构建的formatted_history。1. 确保save_memory_node被正确调用且无报错。2. 确认chat_history在状态中正确传递和更新。3. 检查MessagesPlaceholder和消息列表的构建逻辑。LangGraph 图编译错误。节点函数签名与State定义不匹配或边定义有循环。仔细检查State的TypedDict定义和每个节点函数的输入输出。确保所有节点函数都接收并返回GraphState类型或兼容的字典。使用print调试每个节点的输入输出。4.2 性能与效果优化方向检索优化RAG 核心混合检索 结合向量检索语义相似和关键词检索BM25提高召回率。重排序Re-ranking 使用交叉编码器模型对检索出的文档进行精排将最相关的放在前面。元数据过滤 在向量检索时加入文档来源、章节等元数据过滤条件。提示词工程少样本Few-Shot提示 在系统提示词中提供几个高质量的问答示例引导模型输出格式和风格。思维链Chain-of-Thought 对于复杂推理问题提示模型先一步步思考再给出最终答案。引入 LangSmith这是提升开发效率的“神器”。通过 LangSmith你可以可视化整个工作流的执行轨迹查看每个节点的输入输出对不同的提示词或检索策略进行效果评估和对比快速定位问题所在。模型微调如果通用模型在金融领域的术语、逻辑或格式上表现不佳可以考虑使用LoRA等技术对 Qwen 模型进行高效微调使其更贴合专业领域。图工作流复杂化当前是简单的“检索-生成”两段式。可以引入更多节点例如问题分类节点 更精细地路由到不同处理子图如查询股价、解释术语、生成报告。查询改写节点 根据对话历史改写当前问题使其更适合检索。答案验证节点 调用另一个 LLM 或规则引擎对生成的答案进行事实核查。通过以上步骤你不仅构建了一个可运行的金融问答机器人更掌握了一套基于 LangChain 和 LangGraph 构建生产级 AI Agent 的方法论。从明确 LangGraph 的状态图思想到实现记忆、检索等核心模块再到通过 FastAPI 提供服务最后考虑生产部署和优化这条路径适用于大多数需要复杂编排和状态管理的智能体场景。记住可靠的 AI 应用不仅仅是模型调用更是对数据流、状态和业务逻辑的精心编排。