别再到处找数据了!用Python+Tushare Pro免费获取A股行情、财务、资金流数据(附完整代码)
2026/6/4 5:45:55
做 AI Agent 毕设,最爽的时刻是第一次跑通 demo:终端里噼里啪啦输出,LLM 有问有答,仿佛明天就能答辩。然而爽点过后,真正的痛苦才刚开始:
main.py,想改一条 prompt 得全文搜索。结果就是:功能看上去“有”,却不敢动代码;一动就崩,一崩就通宵。效率低到怀疑人生,更别提炼出论文工作量。
| 维度 | LangChain | LlamaIndex | 自研轻量框架 |
|---|---|---|---|
| 学习曲线 | 高,概念多(Chain、Agent、Tool、Memory) | 中,专注索引与检索 | 低,只实现必要抽象 |
| 依赖体积 | 重,20+ 子包 | 中,10+ 子包 | 轻,单文件即可运行 |
| 调试透明性 | 黑盒,报错栈深 | 中,检索链路可追踪 | 白盒,自己写的自己调 |
| 小项目适用性 | 杀鸡用牛刀,冷启动 3s+ | 偏向问答检索场景 | 1s 内启动,毕设够用 |
结论:毕设周期 3-4 个月,人力 1 人,算力 1 张 3060。选“手搓”轻量框架最划算,把 LangChain 的设计思想“借”过来,代码量压到 300 行,既能在答辩时讲清楚,也能在简历上写“自研 Agent 框架”。
思路一句话:把“对话”拆成“任务”,把“任务”丢进队列,让“工具”自己注册自己。
关键三点:
Task对象,不管背后是 LLM 调用还是 Python 函数。Session,自动落盘到sessions/{uuid}.json,重启可续跑。@tool(name, desc)一键注册,参数模型用 Pydantic 保证幂等性,同一输入多次执行结果不变,方便缓存。以下代码全部单文件可跑,Python≥3.9,仅依赖openai、pydantic、rich。
复制到mini_agent.py,python 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( 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 decoratorclass 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))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}"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()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())跑通后,你会得到:
sessions/{uuid}.json——对话历史,可回放。result.txt——工具写出的文件。openai.Model.list()做懒热身,把延迟从 3s 降到 500ms。Agent.llm加令牌桶限速,每秒 ≤3 次,超量则本地缓存命中或返回“请求过于频繁”。pydantic自动校验 + 正则黑名单,拦截 SQL 注入、路径穿越等恶意输入,毕设答辩也能讲“安全考量”。tenacity重试 + 指数退避,最大 5 次。rich控制台外,再写一份结构化日志到log.ndjson,方便 Grafana 可视化,老师一看就觉得“工业级”。把上面的模板跑通后,你其实已经拥有了一个可扩展、可持久化、可缓存的 Agent 内核。下一步不妨思考:
毕业设计不是“堆功能”,而是“做权衡”。先让系统跑得快,再让它跑得巧。动手把你之前的“单体脚本”重构成模块化 Agent,你会发现:原来 50% 的代码真的可以删掉,而剩下的 50% 变得无比清晰。祝你答辩顺利,也祝这段轻量代码成为你未来更大项目的种子。