从 0 到 1 搭建 NexusAgent

企业知识库与流程自动化 Agent 项目的完整实践

项目：NexusAgent｜技术栈：FastAPI、DeepSeek、LangGraph、PostgreSQL、原生前端

1. 项目背景与最终目标

NexusAgent 的出发点，是把企业里分散的资料、问答需求和流程动作连接起来。很多知识库产品只解决“搜资料”的问题，很多聊天机器人只解决“自然语言回答”的问题，但真实业务里常常还需要执行动作：查询数据、计算结果、生成图表、保留来源、记录过程，并且让不同用户的历史对话互相隔离。

• 知识库能力：上传 TXT、Markdown、PDF 后自动解析、分块并进入检索。
• 问答能力：基于来源片段生成答案，命中不足时明确拒答。
• Agent 能力：根据用户意图自动规划工具并组合知识库结果。

2. 整体架构与技术选型

项目采用前后端分离但部署上保持轻量的结构。后端使用 FastAPI 统一提供认证、文档、知识库、对话和 Agent 接口；前端使用原生 HTML、CSS、JavaScript 构建工作台；大模型能力由 DeepSeek API 提供；任务编排由 LangGraph 承担；PostgreSQL 保存用户与历史对话；文档和知识片段在当前版本中仍使用本地 JSON 文件保存。

FastAPI 入口与接口组织

后端入口集中在 Backend/App.py。它既挂载前端静态资源，也暴露核心 API。这样的结构让本地启动和 Render 部署都比较简单：一个 Uvicorn 服务即可同时承载页面和接口。

源码摘录：Backend/App.py

App = FastAPI(title="NexusAgent", version="0.2.0") App.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) App.mount("/Frontend", StaticFiles(directory=FrontendDirectory), name="Frontend") GeneratedChartsDirectory.mkdir(parents=True, exist_ok=True) App.mount("/GeneratedCharts", StaticFiles(directory=GeneratedChartsDirectory), name="GeneratedCharts") @App.post("/Chat/AgentAsk", response_model=ChatResponse) async def AskWithAgent(Request: ChatRequest, CurrentUser=Depends(RequireCurrentUser)): return await ExecuteAgentAsk(Request, CurrentUser)

3. 知识库与 RAG 问答实现

知识库的第一步是让资料进入系统。NexusAgent 支持 TXT、Markdown 和 PDF。上传后，DocumentService 会根据文件类型抽取文本、做基础清洗，然后按段落进行分块。分块结果会写入 KnowledgeItem，每个片段都保留 SourceId、标题、正文、文档 ID 和 ChunkIndex，方便后续引用。

源码摘录：Backend/DocumentService.py - 文档上传与分块

async def ExtractUpload(self, FileData: UploadFile): FileName = FileData.filename or "UploadedDocument" Suffix = Path(FileName).suffix.lower() if Suffix not in self.SupportedSuffixes: raise HTTPException(status_code=400, detail="仅支持上传 TXT、Markdown 或 PDF 文档。") RawContent = await FileData.read() if not RawContent: raise HTTPException(status_code=400, detail="上传文件为空。") if Suffix == ".pdf": Content = self.ExtractPdfText(RawContent) else: Content = self.DecodeText(RawContent) Content = self.NormalizeText(Content) if not Content: raise HTTPException(status_code=400, detail="未能从文件中解析出文本内容。") return FileName, Content def BuildChunks(self, Content, ChunkSize=1200, Overlap=120): Paragraphs = [Paragraph.strip() for Paragraph in re.split(r"\n\s*\n", Content) if Paragraph.strip()] Chunks = [] Current = "" for Paragraph in Paragraphs: Candidate = f"{Current}\n\n{Paragraph}".strip() if Current else Paragraph if len(Candidate) <= ChunkSize: Current = Candidate continue if Current: Chunks.append(Current) Current = Paragraph while len(Current) > ChunkSize: Chunks.append(Current[:ChunkSize]) Current = Current[ChunkSize - Overlap :] if Current: Chunks.append(Current) return Chunks or [Content[:ChunkSize]]

在检索实现上，项目没有一开始接入复杂向量数据库，而是先实现了一个本地稀疏检索：对中文做 2 到 6 字的 n-gram 扩展，对英文、数字、符号关键词直接保留，再根据标题命中、正文命中、完整查询命中等规则累积分数。这种方案不完美，但非常适合 MVP 阶段快速验证 RAG 链路。

源码摘录：Backend/RuntimeStore.py - 本地检索

def ExtractSearchTerms(self, Query): CleanQuery = Query.lower() Terms = re.findall(r"[a-zA-Z0-9_#+.%/-]+|[\u4e00-\u9fff]+", CleanQuery) ExpandedTerms = [] for Term in Terms: if re.fullmatch(r"[\u4e00-\u9fff]+", Term): ExpandedTerms.append(Term) for Size in (2, 3, 4, 5, 6): ExpandedTerms.extend(Term[Index : Index + Size] for Index in range(0, len(Term) - Size + 1)) else: ExpandedTerms.append(Term) StopTerms = { "我们", "公司", "这个", "那个", # ... 省略非关键实现 ... "进行", "当前", } return [Term for Term in ExpandedTerms if Term and Term not in StopTerms] def SearchKnowledge(self, Query, Limit=4): QueryTerms = self.ExtractSearchTerms(Query) Results = [] for Item in self.KnowledgeItems: SearchText = f"{Item.Title} {Item.Content}".lower() Score = 0 if Query.lower() in SearchText: Score += 24 for Term in QueryTerms: Count = SearchText.count(Term) if Count == 0: continue if Term in Item.Title.lower(): Score += Count * 5 elif len(Term) >= 3: Score += Count * 2 else: Score += Count if Term in Item.VectorTerms: Score += 1 if Score > 0: Snippet = Item.Content[:220] + ("..." if len(Item.Content) > 220 else "") # ... 省略非关键实现 ... ) ) Results.sort(key=lambda Item: Item.Score, reverse=True) return Results[:Limit]

RAG 问答流程的关键不是“让模型回答”，而是让模型基于可追溯的来源回答。当没有命中 Sources 时，服务会直接提示需要先上传相关文档，避免模型编造不存在的依据。

DeepSeek API 调用封装

模型调用集中封装在 DeepSeekService 中。GenerateAnswer 会组装 system prompt、最近对话上下文和本轮用户 prompt，然后请求 DeepSeek 的 /chat/completions 接口。这里也处理了缺少 API Key、请求超时、HTTP 错误和网络错误等兜底逻辑。

源码摘录：Backend/DeepSeekService.py - 调用 DeepSeek API

async def GenerateAnswer(self, SystemPrompt, ConversationMessages, UserPrompt, Sources, ToolCalls): if not self.Settings.DeepSeekApiKey: return self.GenerateLocalFallbackAnswer(Sources, ToolCalls, "当前未配置 DEEPSEEK_API_KEY。") Messages = [{"role": "system", "content": SystemPrompt}] Messages.extend(ConversationMessages[-8:]) Messages.append({"role": "user", "content": UserPrompt}) Payload = { "model": self.Settings.DeepSeekModel, "messages": Messages, "stream": False, "temperature": 0.2, } Headers = { "Authorization": f"Bearer {self.Settings.DeepSeekApiKey}", "Content-Type": "application/json", } try: async with httpx.AsyncClient(timeout=self.Settings.RequestTimeoutSeconds) as Client: Response = await Client.post( f"{self.Settings.DeepSeekBaseUrl}/chat/completions", headers=Headers, json=Payload, ) Response.raise_for_status() Data = Response.json() except httpx.TimeoutException: return self.GenerateLocalFallbackAnswer(Sources, ToolCalls, "DeepSeek 请求超时，已返回本地执行摘要。") except httpx.HTTPStatusError as Error: return self.GenerateLocalFallbackAnswer(Sources, ToolCalls, f"DeepSeek API 返回 {Error.response.status_code}，已返回本地执行摘要。") except httpx.RequestError as Error: Detail = str(Error) or Error.__class__.__name__ return self.GenerateLocalFallbackAnswer(Sources, ToolCalls, f"DeepSeek 网络请求失败：{Detail}。已返回本地执行摘要。") return Data["choices"][0]["message"]["content"].strip()

4. Agent 工作流与工具调用

当项目从 RAG 问答升级到 Agent 自动化时，最大的变化是：系统不再只是检索资料并生成回答，而是需要判断用户意图、决定是否调用工具、执行工具并把结果整合进最终答案。NexusAgent 使用 LangGraph 把这个过程拆成固定节点，既方便调试，也方便前端展示执行轨迹。

源码摘录：Backend/AgentWorkflow.py - LangGraph 编排

class AgentWorkflowState(TypedDict, total=False): Message: str ConversationMessages: List[Dict[str, str]] Intent: str Sources: List[SourceItem] ToolPlans: List[Dict[str, Any]] ToolCalls: List[ToolCallItem] WorkflowSteps: List[WorkflowStepItem] Answer: str class AgentWorkflow: def __init__(self): Graph = StateGraph(AgentWorkflowState) Graph.add_node("IdentifyIntent", self.IdentifyIntent) Graph.add_node("RetrieveKnowledge", self.RetrieveKnowledge) Graph.add_node("PlanTools", self.PlanTools) Graph.add_node("ExecuteTools", self.ExecuteTools) Graph.add_node("GenerateAnswer", self.GenerateAnswer) Graph.set_entry_point("IdentifyIntent") Graph.add_edge("IdentifyIntent", "RetrieveKnowledge") Graph.add_edge("RetrieveKnowledge", "PlanTools") Graph.add_edge("PlanTools", "ExecuteTools") Graph.add_edge("ExecuteTools", "GenerateAnswer") Graph.add_edge("GenerateAnswer", END) self.CompiledGraph = Graph.compile() async def Run(self, Message, ConversationMessages): InitialState = { "Message": Message, "ConversationMessages": ConversationMessages, "WorkflowSteps": [], # ... 省略非关键实现 ... def AppendStep(self, State, StepName, Status, Detail): Steps = list(State.get("WorkflowSteps", [])) Steps.append(WorkflowStepItem(StepName=StepName, Status=Status, Detail=Detail)) return Steps

工具规划采用关键词触发的轻量方案。比如用户提到“时间”“几点”会触发当前时间工具，提到“工时”“考勤”“月报”会触发工时统计，提到“销售”“奖金”会触发销售数据与奖金计算，提到“图表”“折线图”则会触发图表生成工具。

源码摘录：Backend/AgentTools.py - 工具规划

def PlanTools(self, Message): LowerMessage = Message.lower() Plans = [] WantsChart = any(Keyword in Message for Keyword in ("画图", "图表", "折线图", "柱形图", "柱状图", "趋势图", "可视化")) if any(Keyword in Message for Keyword in ("时间", "几点", "日期", "今天")): Plans.append({"ToolName": "GetCurrentTime", "Input": {"Timezone": "Asia/Shanghai"}}) if any(Keyword in Message for Keyword in ("工时", "打卡", "考勤", "月报", "出勤", "日报")): Plans.append({"ToolName": "QueryWorkHours", "Input": {"Month": self.ExtractMonth(Message)}}) if any(Keyword in Message for Keyword in ("销售", "奖金", "营收", "业绩", "华东", "华南", "华北")): Region = self.ExtractRegion(Message) Plans.append({"ToolName": "QuerySalesData", "Input": {"Region": Region}}) if "奖金" in Message: Plans.append({"ToolName": "CalculateSalesBonus", "Input": {"Region": Region}}) Expression = self.ExtractMathExpression(Message) if Expression: Plans.append({"ToolName": "SafeCalculate", "Input": {"Expression": Expression}}) if "calculate" in LowerMessage and not Expression: Plans.append({"ToolName": "SafeCalculate", "Input": {"Expression": Message}}) if WantsChart: Plans.append( { "ToolName": "RenderDocumentChart", "Input": { "ChartType": self.ExtractChartType(Message), "DataKind": self.ExtractChartDataKind(Message), "Query": Message, "Month": self.ExtractMonth(Message), }, } ) return Plans

为了避免工具变成不可控的黑盒，项目把每次工具调用都包装成 ToolCallItem，其中包含工具名、输入、输出、状态和错误信息。这样前端可以把工具执行结果展示给用户，后端也可以把失败原因记录进运行日志。

源码摘录：Backend/AgentTools.py - 安全计算与图表生成

def SafeCalculate(self, Expression): if not Expression or len(Expression) > 120: raise ValueError("数学表达式为空或过长。") ParsedExpression = ast.parse(Expression, mode="eval") return {"Expression": Expression, "Result": self.EvaluateAst(ParsedExpression.body)} def EvaluateAst(self, Node): if isinstance(Node, ast.Constant) and isinstance(Node.value, (int, float)): return Node.value if isinstance(Node, ast.BinOp) and type(Node.op) in self.AllowedOperators: Left = self.EvaluateAst(Node.left) Right = self.EvaluateAst(Node.right) if isinstance(Node.op, ast.Pow) and abs(Right) > 8: raise ValueError("指数过大，已拒绝执行。") return self.AllowedOperators[type(Node.op)](Left, Right) if isinstance(Node, ast.UnaryOp) and type(Node.op) in self.AllowedOperators: return self.AllowedOperators[type(Node.op)](self.EvaluateAst(Node.operand)) raise ValueError("表达式包含不允许的内容。") def RenderDocumentChart(self, ChartType, DataKind, Query, Month): Series = self.BuildChartSeries(ChartType, DataKind, Query, Month) if len(Series["Points"]) < 2: raise ValueError("未在已上传文档中找到足够的数值数据，至少需要两个数据点才能画图。") FileName = f"Chart-{uuid4().hex[:12]}.svg" GeneratedChartsDirectory.mkdir(parents=True, exist_ok=True) ChartPath = GeneratedChartsDirectory / FileName ChartPath.write_text( self.BuildChartSvg( Series["Title"], Series["Points"], Series["ChartType"], Series["XLabel"], Series["YLabel"], ), encoding="utf-8", ) # ... 省略非关键实现 ... "ChartUrl": ChartUrl, "ChartMarkdown": f"![{Series['Title']}]({ChartUrl})", "Source": Series["Source"], }

5. 用户系统、历史对话与运行日志

企业级 Agent 和个人 Demo 最大的区别之一，是需要区分用户、保留上下文并支持审计。NexusAgent 增加了注册、登录和 Token 会话机制，并把用户、对话和消息写入 PostgreSQL。这样同一个部署实例可以服务多个用户，每个用户只能访问自己的历史对话。

源码摘录：Backend/AuthService.py - 登录凭证与密码校验

def HashPassword(self, Password, Salt): Digest = hashlib.pbkdf2_hmac("sha256", Password.encode("utf-8"), Salt.encode("utf-8"), 120000) return Digest.hex() def BuildAuthResponse(self, User): Token = secrets.token_urlsafe(32) self.Sessions[Token] = { "UserName": User["UserName"], "DisplayName": User["DisplayName"], "CreatedAt": self.GetCurrentTimeStamp(), } return { "Token": Token, "UserName": User["UserName"], "DisplayName": User["DisplayName"], "Status": "Authenticated", } def Login(self, Request): User = self.FindUser(Request.UserName) if not User: raise HTTPException(status_code=401, detail="用户名或密码错误。") ExpectedHash = User.get("PasswordHash", "") ActualHash = self.HashPassword(Request.Password, User.get("PasswordSalt", "")) if not hmac.compare_digest(ExpectedHash, ActualHash): raise HTTPException(status_code=401, detail="用户名或密码错误。") return self.BuildAuthResponse(User)

每次问答还会生成 RunId，记录用户问题、答案预览、来源数量、状态、耗时、Sources、ToolCalls、WorkflowSteps 和错误信息。这个设计让“模型为什么这样回答”不再只能靠猜，而是可以从一次运行记录回溯出来。

源码摘录：Backend/RuntimeStore.py - 对话消息与运行日志

def AppendConversationMessage(self, ConversationId, UserName, Role, Content): Conversation = self.EnsureConversation(ConversationId, UserName) Timestamp = self.GetCurrentTimeStamp() with self.GetConnection() as Connection: SequenceIndex = Connection.execute( """ SELECT COALESCE(MAX("SequenceIndex"), 0) + 1 AS "NextSequenceIndex" FROM "ConversationMessages" WHERE "ConversationId" = %s AND "UserName" = %s """, (ConversationId, UserName), ).fetchone()["NextSequenceIndex"] Connection.execute( """ INSERT INTO "ConversationMessages" ("ConversationId", "UserName", "Role", "Content", "SequenceIndex", "CreatedAt") VALUES (%s, %s, %s, %s, %s, %s) """, (ConversationId, UserName, Role, Content, SequenceIndex, Timestamp), ) Connection.execute( """ UPDATE "Conversations" SET "UpdatedAt" = %s WHERE "ConversationId" = %s AND "UserName" = %s """, (Timestamp, ConversationId, UserName), # ... 省略非关键实现 ... WHERE "ConversationId" = %s AND "UserName" = %s """, (Title, Timestamp, ConversationId, UserName), ) def AddRunLog( self, RunId, ConversationId, UserMessage, Answer, SourceCount, Status, LatencyMs, Sources=None, ToolCalls=None, WorkflowSteps=None, ErrorMessage=None, ): RunLog = RunLogItem( RunId=RunId, ConversationId=ConversationId, UserMessage=UserMessage, AnswerPreview=Answer[:160] + ("..." if len(Answer) > 160 else ""), SourceCount=SourceCount, Status=Status, CreatedAt=self.GetCurrentTimeStamp(), LatencyMs=LatencyMs, Sources=Sources or [], # ... 省略非关键实现 ... self.RunLogs.insert(0, RunLog) self.RunLogs = self.RunLogs[:50] return RunLog

后续优化方向

• 代码由codex生成，结构上还未优化。
• 目前工具数量较少，离真正应用还差得远，只是一次技术学习与测试。
• 目前工具的调用依赖关键词，后续最好是能够实现根据需求动态生成代码并执行。