AI Agent 毕业设计效率提升实战：从单体脚本到可扩展智能体架构-开发者社区

背景痛点：毕业设计里那些“能跑就行”的坑

做 AI Agent 毕设，最爽的时刻是第一次跑通 demo：终端里噼里啪啦输出，LLM 有问有答，仿佛明天就能答辩。然而爽点过后，真正的痛苦才刚开始：

硬编码逻辑像意大利面——prompt、工具调用、结果解析全塞在main.py，想改一条 prompt 得全文搜索。
无状态管理——每次重启脚本，对话上下文全丢，调试一次就要把 20 轮对话重新敲一遍。
同步阻塞调用——LLM、搜索引擎、文件读写全部串行，GPU 空转，人也在工位空转。

结果就是：功能看上去“有”，却不敢动代码；一动就崩，一崩就通宵。效率低到怀疑人生，更别提炼出论文工作量。

技术选型对比：LangChain、LlamaIndex 还是“手搓”轻量框架？

维度	LangChain	LlamaIndex	自研轻量框架
学习曲线	高，概念多（Chain、Agent、Tool、Memory）	中，专注索引与检索	低，只实现必要抽象
依赖体积	重，20+ 子包	中，10+ 子包	轻，单文件即可运行
调试透明性	黑盒，报错栈深	中，检索链路可追踪	白盒，自己写的自己调
小项目适用性	杀鸡用牛刀，冷启动 3s+	偏向问答检索场景	1s 内启动，毕设够用

结论：毕设周期 3-4 个月，人力 1 人，算力 1 张 3060。选“手搓”轻量框架最划算，把 LangChain 的设计思想“借”过来，代码量压到 300 行，既能在答辩时讲清楚，也能在简历上写“自研 Agent 框架”。

核心实现：一张图看懂模块化架构

思路一句话：把“对话”拆成“任务”，把“任务”丢进队列，让“工具”自己注册自己。
关键三点：

任务队列解耦：Agent 主循环只认Task对象，不管背后是 LLM 调用还是 Python 函数。
状态持久化：每轮对话生成一个Session，自动落盘到sessions/{uuid}.json，重启可续跑。
工具调用标准化：装饰器@tool(name, desc)一键注册，参数模型用 Pydantic 保证幂等性，同一输入多次执行结果不变，方便缓存。

代码实战：300 行搞定 Clean Agent

以下代码全部单文件可跑，Python≥3.9，仅依赖openai、pydantic、rich。
复制到mini_agent.py，python mini_agent.py即可体验。

1. 工具注册器与基类

# 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( name=name, description=description, params_model=model Licensing) @functools.wraps(func) async def wrapper(raw_params: dict): validated = model(**raw_params) return await func(validated) return wrapper return decorator

2. 状态管理器：Session + 持久化

class Message(BaseModel): role: str content: str timestamp: datetime = Field(default_factory=datetime.now) class Session(BaseModel): uuid: str = Field(default_factory=lambda: str(uuid.uuid4())) history: List[Message] = [] task_queue: asyncio.Queue = Field(default_factory=asyncio.Queue) def add_message(self, role: str, content: str): self.history.append(Message(role=role, content=content)) def save(self): with open(f"sessions/{self.uuid}.json", "w", encoding="utf-8") as f: # 队列不落地，只存历史 f.write(self.json(exclude={"task_queue"}, ensure_ascii=False))

3. 工具示例：搜索 + 文件写

class SearchParams(BaseModel): query: str @tool(name="search", description="调用 SerpAPI 搜索") async def search_tool(p: SearchParams): # 伪代码，替换成你自己的 SerpAPI 调用 await asyncio.sleep(0.5) return f"Top result for '{p.query}': ..." class WriteParams(BaseModel): filepath: str text: str @tool(name="write", description="写文件") async def write_tool(p: WriteParams): with open(p.filepath, "w", encoding="utf-8") as f: f.write(p.text) return f"Written to {p.filepath}"

4. Agent 主循环：消费任务 + 调用 LLM

import 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( model="gpt-3.5-turbo", messages=[{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": prompt}], temperature=0.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()

5. 快速测试入口

async def main(): import os, shutil os.makedirs("sessions", exist_ok=True) 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())

跑通后，你会得到：