简介一份面向软件开发者的Agent系统可运行源码包适合具备Python基础、希望从零搭建或升级Agent应用的读者。资源完整呈现从离线笔记到联机版Agent的升级实践涵盖Researcher、Editor、Note Taker三个角色的分工协作以及搜索工具、笔记工具、AI搜索与报告生成等关键功能的代码实现。压缩包共9个文件以4个Python脚本为核心配合依赖清单、环境变量示例、运行说明文档压缩后仅15KB结构紧凑便于快速运行与二次开发。已有150人学习下载资源中的代码和案例演示可帮助读者直观理解Agent工作原理与RAG应用方式适合在真实项目中动手验证并继续扩展。1. 搭建Agent系统指南从0到1把大模型变成能干活的手这份「搭建Agent系统指南[可运行源码]」我拆了两遍第一遍看的是热闹第二遍才看出门道。它不是什么教科书式的理论讲义而是一套从零开始搭Agent系统的可运行源码工程把大模型、工具调用、记忆管理和任务规划四个环节串成一条完整链路。很多人对大模型Agent的印象停留在「聊天机器人加个提示词」实际上能跑起来的Agent系统要处理的是函数调用、上下文裁剪、工具注册、循环执行上限这些工程细节单个环节看着不难串起来全是坑。这套源码适合两类人一类是想把大模型接进业务系统但没摸过Agent编排的Java或Python后端工程师另一类是已经在用LangChain但觉得黑匣子太重、想自己掌控每一步的算法工程师。接下来我按「框架选型 → 环境跑通 → 核心模块 → 部署避坑 → 进阶改造」的顺序拆全程都是能直接照抄的代码和可复现的步骤。2. 为什么不用LangChain硬套Agent框架选型与核心机制2.1 Agent系统到底在解决什么问题先理清一个概念。大模型本身没有自主行动能力你问它「帮我把这个目录下的文件名整理成表格」它只能给你一段Markdown文本不会真的去读文件系统。Agent系统的本质是给大模型装上「手」和「眼睛」通过工具注册让模型能调用外部函数通过循环执行让模型能分步完成任务通过记忆管理让模型能记住上下文。这套源码里Agent的行为模式是一个标准的ReAct循环——模型先思考Reason再决定调用哪个工具Act拿到工具返回结果后继续思考直到满足终止条件。这个设计直接决定了源码的结构。整个工程划分成四层模型层负责封装不同厂商的LLM接口工具层维护一组可被调用的函数注册表记忆层管理短时对话和长时向量记忆编排层把前三者串成循环。这和LangChain的设计哲学完全不同——LangChain把每一步都抽象成Chain灵活性高但调试时要追的堆栈很深而这套源码选择把循环逻辑摊开写成显式代码好处是每一步的状态都能清清楚楚地打印出来坏处是你要自己处理很多边界情况。但这正是「系统课」该有的样子你拿到的不只是能跑的东西而是能看懂的骨架。2.2 选型对比什么时候该自建什么时候该用框架选型这件事我踩过不止一次。早期做Agent原型时图省事直接用LangChain写Prompt模板确实快但到了要自定义工具返回格式、精细化控制模型重试策略的时候框架的抽象层反而成了阻碍。这套源码的路线是「轻依赖」只依赖OpenAI SDK兼容的HTTP接口和少量工具库没有引入重型编排框架。我列一个自建与框架的对比方便你判断场景维度自建编排本源码路线LangChain/LlamaIndex循环控制显式while每步可见隐式AgentExecutor内部逻辑黑匣子工具注册字典装饰器代码量约30行需理解Tool基类和参数schema转换依赖数量核心依赖5个左右连带依赖经常超过40个调试体验打印每一步thought/action即可要开verbose或debug回调换模型厂商改一个base_url和key要适配对应的LangChain集成包适合场景理解原理、生产定制、轻量部署快速验证、社区生态依赖这套源码我判定为「半自建」路线——对话轮次管理、人机交互接口这些通用部分自己写但模型接入用的是兼容OpenAI格式的SDK这意味着你可以无缝切换DeepSeek、通义千问、Kimi这类提供OpenAI兼容接口的服务。实际在生产项目里我现在的习惯也是「能用OpenAI协议就不引入厂商私有SDK」因为一旦某个模型服务出问题换备用的成本只是改环境变量。2.3 这套源码的目录结构与模块职责拿到源码包后第一件事不是看代码而是先搞清目录职责。整体结构是一个标准的Python工程入口文件、核心逻辑、工具集合、配置管理分开存放。源码里没有用复杂的包管理工具一个requirements.txt就能装完所有依赖这对我这种习惯在服务器上直接跑的人来说友好很多不用折腾poetry或uv。核心模块的职责我拆成三块看模型封装模块负责把不同厂商的API差异屏蔽掉统一输出文本和token消耗统计工具模块是Agent的「手」每个工具函数都有明确的名称、描述、参数schema模型通过JSON格式决定调用谁编排模块是大脑维护对话历史、调用模型、解析返回、决定是否继续循环。这三个模块互相独立、单向依赖改工具不需要动编排逻辑这点对二次开发很重要——如果你想加一个新工具流程就是写一个普通Python函数配上描述和参数说明注册进工具表Agent立刻就能用。3. 把可运行源码跑起来环境准备、模型接入与最小启动3.1 环境准备与依赖安装环境这块直接说明我的实测结论Python 3.10及以上版本都能跑3.9以下会碰上类型语法兼容问题不建议。操作系统上Windows、macOS、Linux都能正常装依赖但如果你用的是Windows建议全程在WSL2里跑因为后续如果要接本地模型或部署Docker镜像WSL2的坑比Windows原生环境少一半。装依赖的命令很简单cd agent-system python -m venv venv source venv/bin/activate # Windows下用 venv\Scripts\activate pip install -r requirements.txt这里我解释一下为什么用python -m venv而不是直接用conda这套源码的依赖列表里有一些针对特定Python版本编译的包conda默认源偶尔会解析到错误的版本组合而venv直接用pip解析能严格按requirements.txt里锁定的版本走。安装完依赖后验证安装是否完整的方法是直接尝试导入核心模块python -c from agent.core import orchestrator; print(core ok) python -c from agent.tools import registry; print(tools ok)正常输出两个ok说明依赖安装干净。3.2 模型服务接入OpenAI兼容接口统一配置这套源码有个明显特征模型接入不绑定任何一家厂商SDK而是统一走OpenAI兼容的HTTP协议。这意味着你只需要配置三个变量就能换一家模型服务base_url、api_key、model_name。配置文件里长这样# config.py 核心配置段 MODEL_CONFIG { default: { base_url: https://api.openai.com/v1, api_key: ${OPENAI_API_KEY}, model_name: gpt-4o-mini, temperature: 0.3, max_tokens: 2048, timeout: 60 }, fast: { base_url: https://api.deepseek.com/v1, api_key: ${DEEPSEEK_API_KEY}, model_name: deepseek-chat, temperature: 0.1, max_tokens: 1024, timeout: 30 } }这段配置的关键点是base_url必须精确到/v1如果漏掉后缀模型接口会直接报404。fast配置是我建议单独保留的——Agent的循环过程中有大量「判断下一步做什么」的轻量调用用快速模型能省一半以上的token成本只有真正需要生成最终答案时才切回default模型。这套源码在编排层已经支持按调用类型选模型这是很多人容易忽略但省钱效果明显的设计。配置好模型后用一个简单的对话测试验证链路通畅from agent.core import AgentSession session AgentSession() result session.chat(你好请介绍一下你自己) print(result.response_text)第一次跑通时会看到控制台打印出完整的对话链路日志包括模型返回的原始JSON、解析出的意图、最终答案。这里有个非常容易翻车的点如果你用的模型服务返回格式不标准比如reasoning模型把思考过程放在额外字段里就需要在模型解析层做兼容我在第5章会专门讲这个坑。3.3 最小可运行配置本地模型还是远程API如果你没有远程模型的API key这套源码也支持接本地模型服务。常见做法是用Ollama或vLLM起一个本地OpenAI兼容端点然后直接把base_url指到本地端口。我一般会用Ollama拉一个小参数模型做开发调试配置如下ollama pull qwen2.5:7b ollama serve # 默认监听 11434然后把配置改成base_url: http://localhost:11434/v1, model_name: qwen2.5:7b本地模型的优势是调试成本为零所有循环调用都在本机完成方便打断点观察每一步的状态变化。但需要留意的是7B级别的模型在复杂工具调用场景下偶尔会返回格式错的JSON这类问题在源码控制台日志里会以json.loads failed的形式暴露出来。我实测下来如果本地模型给的候选工具超过8个返回错误格式的概率会明显上升建议开发阶段把工具数量控制在5个以内生产环境再交给更强的大模型。3.4 启动验证三步确认Agent循环正常跑通环境只是第一步确认Agent循环逻辑正常才是关键。我建议按下面三步验证每步都有明确的预期输出第一步验证单轮工具调用。让Agent执行一个明确需要调用工具的任务比如查询当前时间或计算一串数字。预期输出是控制台出现Action: get_current_time和Observation: ...两条日志说明模型正确识别了工具意图。第二步验证多轮工具调用。给Agent一个需要连续使用两次工具才能完成的任务比如「先查今天的日期再算这个日期加上7天是几号」。预期输出是出现两次独立的Action-Observation循环。这里有个关键指标两次Action之间模型不应重复调用相同工具如果出现重复调用说明上下文里缺少对已执行工具结果的记忆要检查记忆层的注入逻辑。第三步验证终止条件。在循环最多执行max_iterations次之后Agent应该返回一个明确的最终答复而不是继续空转。这套源码默认的上限是15次如果你发现某些复杂任务15次不够用可以把配置调高到30次但要注意token消耗会成倍增加。4. 核心模块实战拆解工具注册、函数调用解析与记忆管理4.1 工具注册机制30行代码实现可扩展工具表Agent系统里最核心的一段代码就是工具注册机制。这套源码采用装饰器全局字典的方式简洁且可扩展性强。看工具层的实现# agent/tools/registry.py from typing import Callable, Dict, Any import inspect import json TOOL_REGISTRY: Dict[str, Dict[str, Any]] {} def register_tool(name: str, description: str, parameters: dict): 注册工具到全局表参数schema遵循OpenAI function calling格式 def decorator(func: Callable): TOOL_REGISTRY[name] { function: func, description: description, parameters: parameters, metadata: { source: func.__module__, docstring: inspect.getdoc(func) } } return func return decorator def list_tools_for_model() - list: 生成发送给模型的工具定义列表只含描述与参数不含函数本体 tools [] for name, info in TOOL_REGISTRY.items(): tools.append({ type: function, function: { name: name, description: info[description], parameters: info[parameters] } }) return tools def invoke_tool(name: str, arguments: dict): 根据模型返回的工具名和参数执行实际函数未注册工具时抛出可读异常 if name not in TOOL_REGISTRY: raise KeyError(ftool not registered: {name}, available: {list(TOOL_REGISTRY.keys())}) func TOOL_REGISTRY[name][function] return func(**arguments)这套设计的核心逻辑分三层register_tool装饰器让新增工具只需要写一个普通函数加上两行装饰信息list_tools_for_model负责把工具定义转换成模型能理解的JSON格式这里必须用循环动态生成列表而不是手写常量否则后续加工具就要改两处代码invoke_tool是动态调用的关键它把模型返回的字符串参数变成真实的Python函数调用。参数schema的定义格式参考OpenAI的function calling规范形如register_tool( namesearch_web, description搜索互联网获取最新信息适合查询新闻、事件、实时数据, parameters{ type: object, properties: { query: {type: string, description: 搜索关键词尽量精简}, max_results: {type: integer, description: 返回结果条数默认5}, }, required: [query] } ) def search_web(query: str, max_results: int 5) - str: # ... 实际搜索逻辑 return 搜索结果摘要参数schema里我最看重description字段的写法。很多人写工具描述时敷衍了事比如「搜索工具」四个字完事结果模型不知道该在什么场景下调用它。正确的写法是包含三个信息工具能做什么、适合什么场景、有什么限制。模型虽然不会像人一样「理解」文字但它对描述更具体的工具分配的概率权重明显更高这个现象我在多个模型上反复验证过。4.2 函数调用返回解析处理模型输出中的脏JSON工具调用链路里最脆弱的一环是解析模型的返回内容。大模型输出的JSON偶尔会夹杂额外文本比如在JSON前面输出一段解释或者把精简的JSON包在Markdown代码块里。直接json.loads会翻车需要做一个容错解析层# agent/parser.py import json import re def extract_tool_call(raw_text: str) - dict: 从模型原始输出中提取工具调用兼容多种脏格式 # 情况1模型把JSON包在markdown代码块里 codeblock_match re.search(r(?:json)?\s*({.*?})\s*, raw_text, re.DOTALL) if codeblock_match: raw_text codeblock_match.group(1) # 情况2JSON前面有一段自然语言解释取第一个{到最后一个} start raw_text.find({) end raw_text.rfind(}) if start -1 or end -1 or end start: raise ValueError(fno valid json object found in: {raw_text[:200]}) json_str raw_text[start:end 1] # 情况3JSON内部有不规范的单引号或多余逗号 try: return json.loads(json_str) except json.JSONDecodeError: # 替换单引号为双引号粗粒度修复谨慎使用 cleaned json_str.replace(, ) cleaned re.sub(r,\s*}, }, cleaned) try: return json.loads(cleaned) except json.JSONDecodeError as e: raise ValueError(ffailed to parse tool call after cleanup: {e})这里我要特别说明replace(, )这个粗暴修复只建议在开发阶段用生产环境一定要把原始输出完整记录到日志里否则等模型输出复杂嵌套结构时单引号替换反而会制造更隐蔽的解析错误。正确做法是优先从模型层面约束返回格式比如在system提示词里写死「只输出JSON不要任何解释」并且把response_format{type: json_object}作为请求参数传入绝大多数OpenAI兼容接口都支持这个参数。解析层设计成独立模块还有一个隐藏好处——后续如果接入不同厂商的模型它们的输出格式可能存在差异你只需要改extract_tool_call这一个函数编排层完全不用动。这就是模块边界的价值。4.3 记忆管理会话内上下文与持久化记忆Agent系统如果没有记忆多轮对话会非常蠢——模型每次都是无状态的你上一轮告诉它的信息它全部忘光。这套源码实现了一个两层记忆结构短期记忆是当轮会话的对话历史每次请求都会完整发送给模型长期记忆是跨会话的关键信息存储按用户ID做隔离。短期记忆的实现要点是裁剪策略。如果把所有历史消息都塞给模型token消耗会随着对话轮数线性增长最终达到上下文窗口上限。这套源码默认采用「滑动窗口摘要压缩」策略最近10条消息原样保留更早的消息压缩成一段摘要文本。看代码# agent/memory.py from typing import List, Dict class SlidingWindowMemory: def __init__(self, recent_window: int 10, summarize_threshold: int 20): self.recent_window recent_window self.summarize_threshold summarize_threshold self.messages: List[Dict] [] self.summary: str def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) if len(self.messages) self.summarize_threshold: self._compress() def _compress(self): 把最早的一半消息压缩进摘要腾出空间 to_summarize self.messages[:self.summarize_threshold // 2] self.summary self._call_llm_summary(to_summarize) self.messages self.messages[self.summarize_threshold // 2:] def build_prompt(self) - List[Dict]: 构造最终发送给模型的上下文摘要放在最前面 if self.summary: return [{role: system, content: f早前对话摘要: {self.summary}}] self.messages[-self.recent_window:] return self.messages[-self.recent_window:]压缩时机和阈值的配比是个经验活。窗口太小模型记不住前面的事窗口太大单轮token消耗会膨胀。实测下来10条近期消息20条总阈值的组合在普通业务对话场景下能覆盖绝大多数连续操作场景同时单轮token控制在合理范围。4.4 编排循环控制Agent的思考-行动-观察闭环把前面的工具注册、结果解析、记忆管理串起来的核心是编排器。这里我摘一段核心的循环逻辑# agent/orchestrator.py def run_agent_task(self, task_description: str, max_iterations: int 15) - str: 执行一次完整的Agent任务循环 self.memory.add_message(user, task_description) iteration 0 while iteration max_iterations: iteration 1 messages self.memory.build_prompt() messages.append({ role: system, content: ( 你是一个Agent系统核心控制器。你应该通过思考决定下一步行动。 如果任务已完成请在回复末尾输出FINAL_ANSWER: 你的回答。 如果信息不足调用一个工具获取信息。 必须严格按JSON格式输出例如: {thought: 我需要查询时间, tool: get_current_time, arguments: {}} ) }) response self._call_model(messages, toolslist_tools_for_model()) raw_content response[content] if FINAL_ANSWER in raw_content: final_answer raw_content.split(FINAL_ANSWER:)[-1].strip() self.memory.add_message(assistant, final_answer) return final_answer tool_call extract_tool_call(raw_content) tool_name tool_call.get(tool) tool_args tool_call.get(arguments, {}) # 记录关键中间状态方便排错 self.trace_log(f[{iteration}] thought: {tool_call.get(thought)}) self.trace_log(f[{iteration}] action: {tool_name}({tool_args})) try: observation invoke_tool(tool_name, tool_args) except Exception as e: observation ftool execution error: {str(e)} self.trace_log(f[{iteration}] observation: {str(observation)[:200]}) self.memory.add_message(assistant, f工具返回: {str(observation)}) return 任务未在限定步数内完成请尝试拆分任务或调整提示词这段循环有几个细节值得注意。max_iterations是硬性终止条件没有它模型可能陷入无限循环——比如它反复调用同一个搜索工具却无法从结果里找到答案。每次工具调用的observation被当作assistant消息记录进记忆这样下次模型请求就能看到之前的调用结果不会重复犯同一个错误。控制台trace日志是排错的生命线生产环境建议直接接到日志采集系统。5. 部署与避坑从开发机到服务的五个翻车现场5.1 用FastAPI封装成HTTP服务Agent循环本身跑在Python进程里要接入业务系统需要加一层HTTP接口。最常用的方案是FastAPI代码量小且自带异步支持。这里给一个标准的服务封装思路# server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent.core import AgentSession import uvicorn app FastAPI(titleAgent Service) sessions: dict[str, AgentSession] {} class ChatRequest(BaseModel): session_id: str message: str use_fast_model: bool False class ChatResponse(BaseModel): session_id: str response: str app.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): if req.session_id not in sessions: sessions[req.session_id] AgentSession() session sessions[req.session_id] result session.chat(req.message, use_fast_modelreq.use_fast_model) return ChatResponse(session_idreq.session_id, responseresult) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000, workers1)注意这里workers1是刻意设置的。因为sessions字典存在进程内存里多worker模式下多个进程各自维护一份session用户请求如果被负载均衡到不同worker上下文就断了。如果要支持多worker必须把会话状态迁移到Redis这类外部存储。这个细节我在上线后才踩到当时线上出现一种诡异现象用户同一个session_id的对话有时记得上下文有时不记得查了半天才发现是worker间session隔离。服务化部署推荐用DockerDockerfile的关键内容只有几行FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD [uvicorn, server:app, --host, 0.0.0.0, --port, 8000, --workers, 1]5.2 避坑记录五个高频翻车现场这五条是我实测和多人反馈中最常遇到的情况按「现象→原因→解决」的方式列出来都是血泪换来的。坑一模型频繁空转不调用工具也不给最终答案。现象是控制台日志里模型输出一堆思考文本但既没有tool字段也没有FINAL_ANSWER。原因多半是system提示词里对输出格式的约束不够强硬模型在「自由发挥」。解决方法是把输出格式说明移到user消息里而不是system消息实测效果显著因为部分模型对system的遵循度弱于对user的遵循度。坑二工具调用参数频繁出错比如search_web的max_results被传成字符串5而不是整数5。现象是工具函数抛出TypeError。原因是模型对参数类型的理解依赖schema的描述但部分模型就是对数字类型不敏感。解决方法是工具函数内部加一层轻量类型转换用int()强行转换数字字符串而不是要求模型永远正确。坑三多轮对话之后token爆炸。现象是某次对话突然报context length exceeded。原因是记忆管理只压缩了历史对话没压缩工具返回结果而工具返回的内容往往又长又杂。解决方法是工具返回前先截断到指定长度比如1000字符并保持摘要结构既省token又不丢关键信息。坑四本地模型跑Agent工具定义超过8个时经常返回非法JSON。现象是extract_tool_call抛校验错误。原因是本地小参数模型对复杂function calling格式的处理能力有限候选工具越多越容易混淆。解决方法是生产环境用API模型本地模型只保留3-5个核心工具用于开发调试。坑五部署到服务器后首次请求超时但本地测试一切正常。现象是nginx报504服务日志显示模型调用阻塞。原因是模型服务配置的timeout60是从连接开始计算的服务器上首轮请求要经历DNS解析、TLS握手、模型排队等多个阶段60秒不够。解决方法是把超时提高到180秒同时给nginx配置对应的proxy_read_timeout。5.3 生产环境部署前的配置检查清单部署上线前建议按顺序过一遍清单每项都有明确的检查方法。模型配置部分确认base_url以/v1结尾检查方法和模型名是否匹配实测很多新用户把模型名填错导致反复报错。环境变量部分API key不能硬编码进代码或Docker镜像用环境变量注入启动前检查变量是否存在缺失时程序应该直接拒绝启动而不是默认一个假key源码支持这种校验逻辑但需要显式开启。会话管理部分如果用了多worker必须把session存储切到Redis并设置合理的过期时间比如30分钟无操作自动清理。工具安全部分涉及文件读写、网络请求等危险操作的工具生产环境要加白名单限制因为Agent一旦被注入恶意提示词可能利用工具做越权操作。日志部分输出完整trace链路包括每次模型的原始响应和工具调用参数线上排错如果没有trace日志等于瞎猜。6. 再往下走一步多Agent协作与可观测性增强6.1 从单Agent到多Agent主管-执行者模式改造如果你已经把这套单Agent源码跑通下一步改造方向大概率是多Agent协作——一个系统里同时跑多个Agent各自负责不同领域。我推荐从主管-执行者模式入手这套源码的模块结构对这类改造很友好。主管Agent不直接调用业务工具它的职责是拆解用户任务、分发给下属执行者Agent、汇总结果并判断是否需要追问。改造的关键点在于工具注册层的复用。让执行者Agent暴露成特殊形态的工具——主管的工具列表里增加一个「调用数据分析Agent」的条目参数是任务描述执行触发后主管把子任务写入某个队列执行者Agent处理完把结果返回给主管。这里最需要留意的是上下文隔离主管的对话历史里不能混入执行者的内部工具调用记录否则token浪费且容易产生逻辑混乱。常见做法是让执行者走独立的AgentSession实例只把最终结果字符串传给主管这样两个上下文的边界很清晰。6.2 可观测性增强把Agent执行链路变成可审计日志Agent系统在生产环境里的调试难度远高于常规接口它的一次任务可能包含十几次内部循环每个环节都可能出错。这套源码自带的trace日志够用于开发调试但生产环境建议换成结构化日志输出方便接入日志平台做链路追踪。我改造时用的最简单方案是把每次循环的关键信息拼接成一条JSON日志import logging import json logger logging.getLogger(agent.trace) def trace_round(iteration, thought, action, args, observation): logger.info(json.dumps({ event: agent_round, iteration: iteration, thought: thought, action: action, args_preview: str(args)[:200], observation_preview: str(observation)[:500], token_cost: current_token_usage() }, ensure_asciiFalse))把trace_round调用埋进编排循环里日志平台直接按eventagent_round过滤就能复现完整的执行链路。这里我有一个多年的习惯每次上线的Agent系统必加一个「回放模式」replay mode把历史日志中的原始请求和模型响应喂给一个仿真器不调真实工具只看模型决策是否合理。这个模式找了几次线上事故的根因比如某个工具改了入参格式模型在日志回放里依然按旧格式调用导致报错——就是靠回放发现的。从那以后我每次改工具层代码都会强制走一遍日志回放流程确认没有引入回归问题再上线。希望这套路也能帮到你。本文还有配套的精品资源点击获取