核心内容摘要
红桃17·C18：一场关于心动与抉择的浪漫序曲
背景痛点毕业设计里那些“能跑就行”的坑做 AI Agent 毕设最爽的时刻是第一次跑通 demo终端里噼里啪啦输出LLM 有问有答仿佛明天就能答辩。
然而爽点过后真正的痛苦才刚开始硬编码逻辑像意大利面——prompt、工具调用、结果解析全塞在main.py想改一条 prompt 得全文搜索。
无状态管理——每次重启脚本对话上下文全丢调试一次就要把 20 轮对话重新敲一遍。
同步阻塞调用——LLM、搜索引擎、文件读写全部串行GPU 空转人也在工位空转。
结果就是功能看上去“有”却不敢动代码一动就崩一崩就通宵。
效率低到怀疑人生更别提炼出论文工作量。
技术选型对比LangChain、LlamaIndex 还是“手搓”轻量框架维度LangChainLlamaIndex自研轻量框架学习曲线高概念多Chain、Agent、Tool、Memory中专注索引与检索低只实现必要抽象依赖体积重20 子包中10 子包轻单文件即可运行调试透明性黑盒报错栈深中检索链路可追踪白盒自己写的自己调小项目适用性杀鸡用牛刀冷启动 3s偏向问答检索场景1s 内启动毕设够用结论毕设周期
个月人力 1 人算力 1 张 3060。
选“手搓”轻量框架最划算把 LangChain 的设计思想“借”过来代码量压到 300 行既能在答辩时讲清楚也能在简历上写“自研 Agent 框架”。
核心实现一张图看懂模块化架构思路一句话把“对话”拆成“任务”把“任务”丢进队列让“工具”自己注册自己。
关键三点任务队列解耦Agent 主循环只认Task对象不管背后是 LLM 调用还是 Python 函数。
状态持久化每轮对话生成一个Session自动落盘到sessions/{uuid}.json重启可续跑。
工具调用标准化装饰器tool(name, desc)一键注册参数模型用 Pydantic 保证幂等性同一输入多次执行结果不变方便缓存。
代码实战300 行搞定 Clean Agent以下代码全部单文件可跑Python≥
9仅依赖openai、pydantic、rich。
复制到mini_agent.pypython mini_agent.py即可体验。
工具注册器与基类# mini_agent.py import json, time, uuid, asyncio, functools from typing import Any, Dict, List, Callable from pydantic import BaseModel, Field from datetime import datetime from rich.console import Console console Console() class ToolMeta(BaseModel): name: str description: str params_model: type[BaseModel] registered_tools: Dict[str, ToolMeta] {} def tool(name: str, description: str): def decorator(func: Callable): params_model func.__annotations__ # 简化只取第一个参数作为 Pydantic 模型 model params_model[list(params_model.keys())[0]] registered_tools[name] ToolMeta( namename, descriptiondescription, params_modelmodel Licensing) functools.wraps(func) async def wrapper(raw_params: dict): validated model(**raw_params) return await func(validated) return wrapper return decorator
状态管理器Session 持久化class Message(BaseModel): role: str content: str timestamp: datetime Field(default_factorydatetime.now) class Session(BaseModel): uuid: str Field(default_factorylambda: str(uuid.uuid4())) history: List[Message] [] task_queue: asyncio.Queue Field(default_factoryasyncio.Queue) def add_message(self, role: str, content: str): self.history.append(Message(rolerole, contentcontent)) def save(self): with open(fsessions/{self.uuid}.json, w, encodingutf-
as f: # 队列不落地只存历史 f.write(self.json(exclude{task_queue}, ensure_asciiFalse))
工具示例搜索 文件写class SearchParams(BaseModel): query: str tool(namesearch, description调用 SerpAPI 搜索) async def search_tool(p: SearchParams): # 伪代码替换成你自己的 SerpAPI 调用 await asyncio.sleep(
0.
return fTop result for {p.query}: ... class WriteParams(BaseModel): filepath: str text: str tool(namewrite, description写文件) async def write_tool(p: WriteParams): with open(p.filepath, w, encodingutf-
as f: f.write(p.text) return fWritten to {p.filepath}
Agent 主循环消费任务 调用 LLMimport openai openai.api_key sk-xxx class Agent: def __init__(self, session: Session): self.session session async def llm(self, prompt: str) - str: resp await openai.ChatCompletion.acreate( modelgpt-
5-turbo, messages[{role: system, content: You are a helpful assistant.}, {role: user, content: prompt}], temperature
3 ) return resp.choices[0].message.content async def run(self): while not self.session.task_queue.empty(): task: Dict # {type: tool, name: search, params: {...}} task await self.session.task_queue.get() if task[type] tool: meta registered_tools[task[name]] result await meta.params_model(task[params]) self.session.add_message(tool, result) elif task[type] llm: answer await self.llm(task[prompt]) self.session.add_message(assistant, answer) self.session.save()
快速测试入口async def main(): import os, shutil os.makedirs(sessions, exist_okTrue) session Session() session.add_message(user, 请搜索 AI Agent 毕业设计 关键词并把摘要写到 result.txt) await session.task_queue.put({type: tool, name: search, params: {query: AI Agent 毕业设计}}) await session.task_queue.put({type: tool, name: write, params: {filepath: result.txt, text: 搜索得到的摘要...}}) agent Agent(session) await agent.run() console.print([bold green]Done! check result.txt) if __name__ __main__: asyncio.run(main())跑通后你会得到sessions/{uuid}.json——对话历史可回放。
result.txt——工具写出的文件。
控制台实时输出方便打断点。
性能与安全让 demo 级代码也能“见人”冷启动延迟LLM 首次调用要加载 tokenizer可提前openai.Model.list()做懒热身把延迟从 3s 降到 500ms。
LLM 调用频次控制在Agent.llm加令牌桶限速每秒 ≤3 次超量则本地缓存命中或返回“请求过于频繁”。
输入过滤过滤用pydantic自动校验 正则黑名单拦截 SQL 注入、路径穿越等恶意输入毕设答辩也能讲“安全考量”。
生产环境避坑指南别让“彩蛋”变成“雷弹”无限递归Agent 自己给自己发任务时一定加深度计数器 ≥5 即强制退出。
API 限流OpenAI 连续 429 报错时用tenacity重试 指数退避最大 5 次。
日志追踪除rich控制台外再写一份结构化日志到log.ndjson方便 Grafana 可视化老师一看就觉得“工业级”。
结尾算力有限复杂度与速度如何权衡把上面的模板跑通后你其实已经拥有了一个可扩展、可持久化、可缓存的 Agent 内核。
下一步不妨思考模型越大越聪明可 7B 模型在笔记本上跑 2token/s是否够用工具链越丰富越强大可每多一次 LLM 调用就多 1s 延迟能否用本地函数替代状态维护越细粒度越精准可磁盘 IO 与内存同步拖慢整体能否用 LRU 缓存折毕业设计不是“堆功能”而是“做权衡”。
先让系统跑得快再让它跑得巧。
动手把你之前的“单体脚本”重构成模块化 Agent你会发现原来 50% 的代码真的可以删掉而剩下的 50% 变得无比清晰。
祝你答辩顺利也祝这段轻量代码成为你未来更大项目的种子。