ai agent开发，近两年非常热门的话题。目前，ai已经渗透到人类生活中的各个角落，自己尝试开发出一套好用的ai，让ai真正成为你的超级助手，amazing!
1. 基础Agent循环框架（思考-行动-观察闭环）
最核心执行逻辑，自主循环执行任务
def simple_agent(task, tools): memory = [] while True: # 1.思考 prompt = f"历史:{memory},任务:{task},可选工具:{tools}" thought = llm_call(prompt) memory.append(f"思考：{thought}") if "完成" in thought: return memory # 2.调用工具 action = parse_action(thought) obs = run_tool(action) # 3.观察存入记忆 memory.append(f"工具返回：{obs}")# 模拟接口def llm_call(p): return "调用查询工具"def parse_action(t): return "search"def run_tool(a): return "查询结果"
2. 工具调用封装（Function Calling标准模板）
对接大模型函数调用，自动构造工具描述、解析返回
import jsondef get_tool_schema(): tools = [ { "type": "function", "function": { "name": "calculator", "description": "数学计算", "parameters": { "type": "object", "properties": { "exp": {"type": "string", "description": "计算公式"} }, "required": ["exp"] } } } ] return toolsdef calculator(exp): return eval(exp)def parse_tool_call(res): args = json.loads(res["function_call"]["arguments"]) return calculator(args["exp"])
3. 简易RAG检索记忆（短期知识库）
Agent外挂文档检索，解决上下文长度限制
from sklearn.feature_extraction.text import TfidfVectorizerimport numpy as npclass ShortMemory: def __init__(self): self.docs = [] self.vec = TfidfVectorizer() def add(self, text): self.docs.append(text) def search(self, query, top_k=2): tf = self.vec.fit_transform(self.docs + [query]) q_vec = tf[-1] doc_vec = tf[:-1] sim = np.dot(doc_vec, q_vec.T).toarray().flatten() idx = np.argsort(-sim)[:top_k] return [self.docs[i] for i in idx]mem = ShortMemory()mem.add("AI Agent依靠大模型+工具实现自主任务")print(mem.search("智能体原理"))
4. 状态机Agent（LangGraph极简复刻）
用状态拆分规划、执行、校验节点，防止任务跑偏
class StateAgent: def __init__(self, task): self.state = {"task": task, "plan": [], "result": None, "step": 0} def plan(self): self.state["plan"] = ["检索资料", "计算", "汇总"] self.state["step"] = 1 def execute(self): if self.state["step"] == 1: self.state["step"] = 2 return "资料已获取" elif self.state["step"] == 2: self.state["result"] = "计算完成" self.state["step"] = 3 def run(self): self.plan() while self.state["step"] < 3: self.execute() return self.state["result"]agent = StateAgent("统计数据")print(agent.run())
5. 多轮对话记忆管理（区分短期/长期）
自动拼接历史对话，控制上下文长度防溢出
class ChatMemory: def __init__(self, max_len=6): self.msg = [] self.max_len = max_len def add(self, role, content): self.msg.append({"role": role, "content": content}) # 超出长度丢弃最早记录 if len(self.msg) > self.max_len: self.msg = self.msg[-self.max_len:] def get_history(self): return self.msgmem = ChatMemory()mem.add("user", "帮我算1+2")mem.add("assistant", "3")print(mem.get_history())
6. 输出格式校验器（规避幻觉、格式错乱）
强制Agent输出JSON，不合法自动重试
import jsondef safe_parse_json(text): try: start = text.find("{") end = text.rfind("}") + 1 return json.loads(text[start:end]) except Exception as e: return Nonedef agent_qa(prompt): resp = llm_call(prompt + "仅输出JSON，key:answer") data = safe_parse_json(resp) if not data: return agent_qa(prompt + "格式错误，重新输出标准JSON") return data["answer"]def llm_call(p): return '{"answer":"结果"}'
7. 多Agent协同调度（主Agent分发子任务）
主智能体拆分任务，调用多个子Agent并行处理
class SubAgent: def __init__(self, name): self.name = name def run(self, sub_task): return f"{self.name}完成：{sub_task}"class MasterAgent: def __init__(self): self.workers = [SubAgent("检索Agent"), SubAgent("计算Agent")] def split_task(self, task): return ["查询数据", "统计求和"] def run(self, task): subs = self.split_task(task) res = [w.run(subs[i]) for i, w in enumerate(self.workers)] return "汇总：" + "；".join(res)master = MasterAgent()print(master.run("月度数据统计"))
8. 工具调用限流+沙箱安全封装
限制调用次数，防止无限循环调用工具
class SafeToolBox: def __init__(self, max_call=5): self.count = 0 self.max = max_call def call(self, func, *args): if self.count >= self.max: return "调用次数超限，停止执行" self.count += 1 return func(*args)def add(a,b): return a+btool = SafeToolBox()print(tool.call(add, 2,3))
9. 强化反馈回路（自动校验结果，错误回滚重跑）
执行后自动校验，不达标重新规划执行
def check_result(output): # 业务校验规则 return isinstance(output, int) and output > 0def feedback_agent(task): for _ in range(3): res = execute_task(task) if check_result(res): return f"校验通过：{res}" task = f"上次结果错误{res}，重新规划" return "多次执行失败"def execute_task(t): return 10print(feedback_agent("求正数"))
10. 简易Agent日志监控（全链路可观测）
记录每一步思考、工具、结果，方便调试
import timeclass AgentLogger: def __init__(self): self.logs = [] def record(self, stage, content): self.logs.append({ "time": time.time(), "stage": stage, "content": content }) def show(self): for item in self.logs: print(f"[{item['stage']}] {item['content']}")log = AgentLogger()log.record("思考", "需要调用计算器")log.record("工具", "calc(10+20)")log.record("结果", "30")log.show()
