Langchain-Chatchat API接口调用示例与封装方法

Langchain-Chatchat API 接口调用与封装实战指南

在企业智能化升级的浪潮中，如何让大模型真正“懂”你的业务，而不是泛泛而谈？一个常见的挑战是：尽管市面上的大语言模型知识渊博，但它们对企业内部制度、产品手册或技术文档一无所知。更关键的是，把敏感资料上传到云端服务存在合规风险。

这正是本地知识库问答系统大显身手的场景。Langchain-Chatchat作为开源社区中的明星项目，凭借其全链路本地化、高可集成性的特点，成为许多团队构建私有AI助手的首选。它不依赖云API，所有数据处理都在内网完成，同时又能提供接近主流大模型的交互体验。

要将这套能力嵌入企业系统——比如OA、客服平台或内部Wiki——核心在于掌握它的API调用方式。本文不会堆砌概念，而是带你从零开始，一步步实现对 Langchain-Chatchat 的高效调用与封装，最终形成一个即插即用的客户端工具类。

我们先来看整个系统的运作逻辑。当你向系统提问“年假有多少天？”时，背后其实经历了一连串精密协作：

首先，你上传的《员工手册》PDF早已被拆解成若干文本块，并通过 BGE 或 Sentence-BERT 这类嵌入模型转化为向量，存入 FAISS 或 Chroma 这样的本地向量数据库。当问题到来，系统同样将其编码为向量，然后在库中快速检索语义最相近的几个段落。这些内容会被拼接进一段精心设计的提示词（prompt），例如：“请根据以下规定回答问题：{检索结果}\n问题：年假有多少天？”最后，这个完整的上下文被送入本地部署的 Qwen 或 ChatGLM3 模型，生成准确且有据可依的回答。

整个流程无需微调模型，却显著降低了“胡说八道”的概率，这就是典型的 RAG（检索增强生成）架构的魅力所在。而这一切能力，都通过一套简洁的 RESTful API 暴露出来，供外部程序调用。

默认情况下，Langchain-Chatchat 的 API 服务运行在http://localhost:8000，基于 FastAPI 框架构建。这意味着你可以像调用 OpenAI 接口一样，用标准的 HTTP 请求与之交互。主要接口包括/chat/completions用于对话，/knowledge_base/upload_file用于知识入库，以及/models查询当前可用模型等。

以最常用的聊天接口为例，请求体是一个 JSON 对象：

{ "model": "qwen", "messages": [ {"role": "user", "content": "什么是Langchain-Chatchat？"} ], "stream": false, "temperature": 0.7, "max_tokens": 1024 }

其中messages字段支持多轮对话，按角色组织历史消息；stream控制是否启用流式输出，对于网页端实时显示回复非常有用；temperature则调节回答的创造性，数值越低输出越稳定。这种设计几乎完全兼容 OpenAI 格式，极大降低了迁移成本。

下面是最基础的同步调用示例。使用 Python 的requests库，我们可以轻松发起一次问答请求：

import requests import json def chat_with_chatchat(question: str, model="qwen"): url = "http://localhost:8000/chat/completions" payload = { "model": model, "messages": [{"role": "user", "content": question}], "stream": False, "temperature": 0.5, "max_tokens": 512 } headers = { "Content-Type": "application/json" } try: response = requests.post(url, data=json.dumps(payload), headers=headers) response.raise_for_status() result = response.json() answer = result['choices'][0]['message']['content'] return answer except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None # 使用示例 if __name__ == "__main__": reply = chat_with_chatchat("请简述Langchain-Chatchat的主要功能") print("AI回复:", reply)

这段代码虽然简单，但包含了生产环境调用的关键要素：JSON 序列化、HTTP 头设置、状态码校验和异常捕获。特别是response.raise_for_status()能自动抛出网络错误，避免因服务未启动或超时导致程序崩溃。

然而，在实际应用中，用户往往希望看到“逐字输出”的效果，就像真人打字一样。这就需要用到流式接口（streaming）。启用stream=True后，服务器会以text/event-stream格式分块返回响应，每一块对应模型生成的一个片段。

以下是实现流式读取的完整代码：

import requests import json def stream_chat(question: str): url = "http://localhost:8000/chat/completions" payload = { "model": "chatglm3", "messages": [{"role": "user", "content": question}], "stream": True, "temperature": 0.1 } headers = { "Content-Type": "application/json" } with requests.post(url, json=payload, headers=headers, stream=True) as r: for line in r.iter_lines(): if line: line_str = line.decode('utf-8') if line_str.startswith("data:"): data = line_str[5:].strip() if data != "[DONE]": try: chunk = json.loads(data) content = chunk['choices'][0]['delta'].get('content', '') if content: print(content, end="", flush=True) except json.JSONDecodeError: continue print() # 换行

这里的关键是设置stream=True并使用iter_lines()逐行解析。每一行以data:开头，我们需要剥离前缀并进行 JSON 解析。注意判断[DONE]标志位以识别流结束。flush=True确保内容立即输出，不会被缓冲。

如果你正在开发一个长期维护的项目，直接写函数显然不够优雅。更好的做法是封装成类，统一管理配置、会话和错误处理。以下是一个实用的客户端封装：

class ChatchatClient: def __init__(self, base_url="http://localhost:8000"): self.base_url = base_url.rstrip('/') self.session = requests.Session() def chat(self, msg: str, model=None, stream=False) -> str or None: url = f"{self.base_url}/chat/completions" payload = { "model": model or "qwen", "messages": [{"role": "user", "content": msg}], "stream": stream, "temperature": 0.5 } try: resp = self.session.post(url, json=payload) resp.raise_for_status() data = resp.json() return data['choices'][0]['message']['content'] except Exception as e: print(f"[Error] 调用失败: {e}") return None def upload_to_knowledge_base(self, file_path: str, kb_name: str): url = f"{self.base_url}/knowledge_base/upload_file" with open(file_path, 'rb') as f: files = {'file': (file_path.split('/')[-1], f, 'application/octet-stream')} data = {'knowledge_base_name': kb_name, 'override': True} try: resp = self.session.post(url, files=files, data=data) resp.raise_for_status() return resp.json() except Exception as e: print(f"[Error] 文件上传失败: {e}") return None # 使用示例 client = ChatchatClient() answer = client.chat("如何部署Langchain-Chatchat？") print(answer) result = client.upload_to_knowledge_base("./docs/manual.pdf", "help_kb") print("上传结果:", result)

这个ChatchatClient类做了几件重要的事：使用requests.Session()复用 TCP 连接，提升并发性能；集中处理异常，避免重复代码；将 URL 拼接逻辑封装在内部，对外暴露简洁接口。未来若需添加认证 Token 或日志记录，也只需在类中扩展即可。

在真实部署环境中，还有一些工程细节值得留意。首先是性能调优。文本分块大小（chunk_size）直接影响检索质量——太小可能导致上下文断裂，太大则引入无关噪声。实践中建议从 256~512 字符起步，结合业务文档特性调整。其次，若服务器配备 GPU，务必启用 CUDA 加速，否则 Embedding 和 LLM 推理速度会成为瓶颈。

安全性方面，虽然系统本身运行在内网，但仍建议增加 API 认证机制。可以在 Nginx 反向代理层添加 JWT 验证，或在 FastAPI 中间件里实现简单的 token 校验。对于高频查询，引入 Redis 缓存能有效减轻后端压力，尤其是那些反复出现的问题。

可观测性也不容忽视。记录每次调用的输入、输出及耗时，不仅能帮助评估系统表现，还能用于后续的知识库优化。例如，当发现某类问题回答不准时，可以检查检索到的上下文是否相关，进而调整分块策略或更换 Embedding 模型。

从架构上看，典型的集成模式是前后端分离：前端负责展示界面，后端通过 ChatchatClient 与 Langchain-Chatchat 通信。知识库更新则可通过定时任务或手动触发完成。整个流程闭环运行，数据不出内网，既满足合规要求，又实现了智能化升级。

回过头看，Langchain-Chatchat 的真正价值不仅在于技术先进性，更在于它为企业提供了“数据主权”。你不再需要为了使用 AI 而牺牲隐私，也不必投入高昂成本训练专属模型。只需几份文档和一次部署，就能拥有一个懂你业务的智能助手。

这种高度模块化的设计思路，正代表着一类新型 AI 应用的发展方向——不是取代人类，而是成为组织记忆的延伸，把散落在各处的知识真正激活起来。