本文面向正在入门LLM Agent的读者先把「Agent 常见模块」在抽象层面讲清楚再用一份可运行的开源样例把概念落到具体文件与调用链上。文中代码与目录均出自示例项目MOSAICMy Own Stories and Ideas Compilation——它是一个隐私优先的本地日记助手只是承载知识点的载体读懂结构之后你可以迁移到任意自己的项目里。编排说明第 110 节专注Agent 各模块如何分工、数据如何流MOSAIC 的前后端形态与怎么跑起来集中在文末附录避免打断主线。文章目录1. 总览两条「入口」与一条「后台线」2. 路由Routing先把用户意图归到「可执行通道」3. 记忆Memory三层「记什么、谁来写、谁来查」4. RAG检索在示例里何时发生、查的是什么5. 工具Tools谁在「声明工具」谁在「调用工具」6. 技能Skills为何用 SKILL.md scripts而不是「全靠模型调脚本」7. 规划Planning何时认为用户一句话里有多步任务8. 多轮对话Multi-turnLLMClient.chat 与 ConversationState 的契约9. 分析型 LLM 调用analyze 作为结构化抽取器10. 小结概念与代码落点速查附录示例应用 MOSAIC 的前后端与使用方式后端单进程 FastAPI前端静态页面 Alpine.jsTelegram 机器人可选第二条入口环境要点速览1. 总览两条「入口」与一条「后台线」从控制流上看示例项目里与 Agent 相关的轴线可以概括为三条与具体业务领域无关换成客服 Bot 或代码工具同理显式命令路由例如 Bot 里的/diary、/ideas、/mood由MainAgent解析命令字与自然语言参数再交给子 Agent。自然语言对话默认走chat由ChatAgent决定是「多步规划执行」还是「带/不带检索的单轮对话」并维护会话状态。与 UI 无关的记忆整理Processor在后台轮询「待处理条目」用 LLM 打标签并写入向量库——可理解为异步的记忆建构/索引不把用户请求阻塞在前台。下面这张图概括主路径省略 HTTP/Telegram 适配层技能执行记忆子系统入口多步 JSON关键词触发analyze embedMainAgent 解析 / 路由ChatAgent 对话与规划SQLite entriesChroma 向量diary 脚本ideas 脚本mood 脚本Plannersearch_memoriesProcessor 后台循环理解这张图后后文每个「Agent 模块」都可以对号入座谁触发、读什么记忆、会不会调工具、产出写回哪里。2. 路由Routing先把用户意图归到「可执行通道」在 Agent 语境里路由负责把开放域输入映射到有限个 handler协议命令、子 Agent、或技能。示例里在MainAgent用极简规则以/开头的文本拆成command args否则一律视为聊天交给ChatAgent。def _parse(self, user_input: str) - tuple: # strip leading slash if present (e.g. /diary 3月11号) text user_input.strip() if text.startswith(/): parts text[1:].split(maxsplit1) command parts[0].lower() args parts[1] if len(parts) 1 else else: command chat args text return command, args def _route(self, command: str, args: str) - str: if command mood: days self._extract_days_for_mood(args) return self.mood_agent.run({days: days, args: args}) if command chat: return self.chat_agent.run({args: args}) date_str self._extract_date(args) routes { diary: self.diary_agent, ideas: self.ideas_agent, } agent routes.get(command) if agent: return agent.run({date: date_str, args: args}) return Unknown command. Send /start to see available commands.读代码时的要点这里是确定性的字符串路由不靠大模型——适合本地小模型场景先收窄问题空间再让 LLM 做「该通道内」的生成或规划。mood与chat单独分支日期类技能共用_extract_date体现「同一类参数解析复用」。子 Agent如DiaryAgent往往只是薄封装真正编排数据与提示词的是skills/*/scripts/generate.pyclass DiaryAgent(BaseAgent): def __init__(self, llm: LLMClient): self.llm llm def run(self, params: dict) - str: date_str params.get(date, None) return generate_diary(date_str)也就是说路由层解「去哪」技能脚本解「怎么做」。3. 记忆Memory三层「记什么、谁来写、谁来查」Agent 的 Memory 常分为短期对话上下文与长期可检索知识。在示例仓库里对应关系很清晰层次载体写入时机读取方式原始事件日志SQLiteentries用户输入即插入可先标为pending按日期/ID SQL 查询对话短期记忆ConversationState.history每轮 user/assistant 追加带长度裁剪作为LLMClient.chat(..., history...)的消息列表可检索长期片段Chroma 集合mosaic_vecProcessor在分析完一条 entry 后add_entryRAGClient.search→get_entries_by_ids结构化原始记忆的表头在init_db中定义——后续 RAG、按日生成、情绪统计都依赖这些列cursor.execute( CREATE TABLE IF NOT EXISTS entries ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, date TEXT NOT NULL, content TEXT NOT NULL, entry_type TEXT DEFAULT NULL, emotion TEXT DEFAULT NULL, source TEXT DEFAULT NULL ) )短期记忆用 dataclass 显式限制轮数避免本地模型被超长历史拖垮dataclass class ConversationState: Lightweight in-memory conversation state. The stored history format must match LLMClient.chat(): history: List[{role: ..., content: ...}]. max_turns: int 10 history: List[Dict[str, str]] field(default_factorylist) def add(self, role: Role, content: str) - None: Add one message to history, then trim to the configured limit. if not isinstance(content, str): content str(content) self.history.append({role: role, content: content}) self._trim()后台写长期向量记忆的路径Processor拉 pending → LLManalyze出emotion/entry_type→ 更新行 → 再 embedding 入库def _process_entries(self): entries get_pending_entries() for entry in entries: try: result self.llm.analyze(entry[3]) update_entry_labels(entry[0], result[emotion], result[entry_type]) self.rag.add_entry({id: entry[0], content: entry[3]}) except Exception as e: print(f[Processor] Failed to process entry {entry[0]}: {e}, will retry later.) continue可以把Processor看成不设用户界面的记忆子 Agent专职「消化」新原子事件让前台对话与生成技能面对的不是一堆pending而是已分类、已向量化的事实。4. RAG检索在示例里何时发生、查的是什么在 Agent 语境里RAG 用查询编码 → 在向量库做近邻检索 → 把片段注入提示词。示例中的RAGClient用 Ollama 算 embeddingChroma 持久化class RAGClient: def __init__(self, embedding_model: str shaw/dmeta-embedding-zh): self.embedding_model embedding_model self.rag_client chromadb.PersistentClient(pathstr(VEC_DB_PATH)) # vector database client self.collection self.rag_client.get_or_create_collection(namemosaic_vec) def add_entry(self, entry: dict): # a single entry Args: entry: dict {id: int, content: str}, the entry to add. response ollama.embeddings( modelself.embedding_model, promptentry[content] ) self.collection.add( ids[str(entry[id])], embeddings response[embedding] )对话侧并不是「每条都 RAG」。ChatAgent用启发式关键词_needs_rag判断是否与「个人记录」相关再调用search_memories内部rag.search 按 ID 回填完整行def _chat(self, user_input: str) - str: # ... history self.state.get().copy() if self._needs_rag(user_input): memories search_memories(user_input, n_results3) # ... 格式化为多条 bullet 片段 ... query ( f{self._SYSTEM_PROMPT}\n fRelevant memories:\n{memories_text}\n\n fUser message:\n{user_input} ) else: query f{self._SYSTEM_PROMPT}\nUser message:\n{user_input} assistant_text self.llm.chat(query, historyhistory) self.state.add(user, user_input) self.state.add(assistant, assistant_text) return assistant_text读代码时的要点RAG 触发是规则 可选检索不是端到端黑模型自决便于本地部署时控制延迟与噪声。search_memories在core/tools.py里与get_current_time、get_weather并列注册进TOOLS概念上同属可被编排可调用的能力对话路径是直接函数调用而非 LLM 自主 function-call JSON。5. 工具Tools谁在「声明工具」谁在「调用工具」经典 Agent 框架里常见两种模式模型产 structured tool call或代码根据流程显式调函数。本示例以后者为主。工具清单在TOOLS字典中汇总def get_current_time() - dict: now datetime.now() return { date: now.strftime(%Y-%m-%d), time: now.strftime(%H:%M), weekday: now.strftime(%A), } def get_weather(city: str Shenzhen) - str: try: response requests.get( fhttps://wttr.in/{city}?format3, timeout5 ) return response.text.strip() except Exception: return Weather unavailable def search_memories(query: str, n_results: int 2) - list: rag RAGClient() ids rag.search(query, n_resultsn_results) entries get_entries_by_ids(ids) return entries TOOLS { get_current_time: get_current_time, get_weather: get_weather, search_memories: search_memories, }在技能层如日记生成Python 直接调用get_current_time、get_weather拼进上下文再把SKILL.md全文与当日记录一起交给 LLM——工具执行发生在提示词组装之前def generate_diary(date_str: str None) - str: # resolve date today datetime.now().strftime(%Y-%m-%d) if date_str is None: time_info get_current_time() date_str time_info[date] # get weather only for today if date_str today: weather get_weather() else: weather Weather unavailable for historical dates # ... skill_prompt load_skill_prompt() full_prompt f{skill_prompt} ## Context - Date: {date_str} - Weather: {weather} ...对照 Agent 概念这里的「工具」不负责推理负责补全模型不知或不可靠的事实当前日期、当日天气SKILL.md里用文字声明工具语义见skills/diary/SKILL.md的Tools小节与代码真实调用形成文档与实现的一对一关系。6. 技能Skills为何用SKILL.md scripts而不是「全靠模型调脚本」示例仓库的 README 描述参考 Agent Skills 思路但针对本地小模型采用两阶段先路由/规划到技能名再由Python 脚本读SKILL.md作为系统级指令去调 LLM。这一结构的结果是路由/规划产出的skill字段是离散、可校验的diary/ideas/mood。生成质量与约束主要由SKILL.md中的 Goal、Instructions、Output Format 锁定。副作用写文件、查库、拉天气放在脚本里避免弱模型在长链条上「假装执行」。planning_prompt.py把「可供规划的技能」和 JSON 输出格式写死给规划用 LLM这是Planner 与 Skills 之间的契约PLANNING_PROMPT You are a task planner for a personal diary assistant called MOSAIC. Given the users input, decompose it into a list of executable steps. ## Available Skills - diary: generate a diary entry for a specific date - ideas: generate an ideas/TODO list for a specific date - mood: analyze emotional trends over the last N days - chat: answer a question or have a conversation based on users records ... ## Output Format [ {{skill: diary, params: {{date: YYYY-MM-DD}}}}, {{skill: mood, params: {{days: 7}}}} ] ...读代码时的要点技能不是「挂在模型上的插件」而是仓库中的一级公民目录 可测试的 Python。这在教学上很友好你把 Agent 拆成「契约prompt 程序script 数据DB/文件」三块分别演进。7. 规划Planning何时认为用户一句话里有多步任务Planner 类做两件事needs_planning用中英关键词启发式判断像不像「然后 / 还有 / and / then」这种多意图粘连句。plan让 LLM 产出严格 JSON 数组每项{ skill, params }再由ChatAgent顺序执行。def needs_planning(self, user_input: str) - bool: Decide if the input looks like a multi-step request. ... text (user_input or ).strip().lower() ... pattern ( |.join(pattern_parts) ) return re.search(pattern, text, flagsre.IGNORECASE) is not None def plan(self, user_input: str) - List[Dict[str, Any]]: Ask the LLM to split the request into a JSON step list. ... today datetime.now().strftime(%Y-%m-%d) prompt PLANNING_PROMPT.format(user_input(user_input or ).strip(), todaytoday) llm_text self.llm.chat(prompt, history[]) # ... JSON parse normalize steps ... return steps执行端在ChatAgent._execute_plan不再经过普通history里的 user 消息重复问题规划路径单独state.add逐步调用generate_diary/generate_ideas/generate_mood最后拼接回复def _execute_plan(self, user_input: str) - str: steps self.planner.plan(user_input) if not steps: return self._chat(user_input) self.state.add(user, user_input) results: List[str] [] for step in steps: ... if skill diary: ... results.append(generate_diary(date_str)) elif skill ideas: ... results.append(generate_ideas(date_str)) elif skill mood: ... results.append(generate_mood(days_int)) else: results.append(f⚠️ Unknown plan step skill: {skill}) merged \n\n.join(results) if results else No plan result produced. self.state.add(assistant, merged) return merged对照 Agent 概念这是典型的Plan-and-Execute雏形——规划与执行在代码中分离规划失败则回退到普通聊天提高鲁棒性。8. 多轮对话Multi-turnLLMClient.chat与ConversationState的契约LLMClient.chat会把当前query追加进传入的history列表再请求 Ollama因此调用方必须小心不要在外部重复追加同一句 user否则模型会看到两遍用户话。ChatAgent在_chat里用快照history self.state.get().copy()生成后再state.add并在注释里写清这一陷阱def chat(self, query, history): ... history.append({role: user, content: query}) response ollama.chat( modelself.model_name, messageshistory) text response.message.content return textdef _chat(self, user_input: str) - str: # IMPORTANT: # LLMClient.chat() appends the provided query into history as a new # {role:user,content:query} message (it mutates the list in-place). # If we first add user_input to state.history and then pass a history # that includes it, the model will see the users input twice. ... history self.state.get().copy()教学意义多轮不仅是「存历史」而是消息拼装顺序与可变引用的工程细节对照示例里这一段能直接映射到你自己写 Agent 时最容易踩的 bug。9. 分析型 LLM 调用analyze作为结构化抽取器除了chatLLMClient.analyze用单独 prompt 要求JSON输出再在 Python 里校验 emotion / entry_type 合法值——这是NLU 或抽取子模块服务于记忆层而非用户可读回复def analyze(self, goal_text): ... full_prompt ANALYZE_ENTRY_PROMPT.format(contentgoal_text) response ollama.chat( modelself.model_name, messages[{role: user, content: full_prompt}]) analyze_result response.message.content json_result self._analyze_result_decode(analyze_result) return json_result它和Processor配合把「非结构化一句用户话」变成数据库可筛、可聚合的字段——后续 mood 统计、按类型拉取 story/idea 都受益于此。10. 小结概念与代码落点速查Agent 概念在示例仓库中的落点可按此顺序读源码路由core/agents/main_agent.py._parse/._route子 AgentDiaryAgent等薄封装 →skills/*/scripts短期记忆core/state.pyLLMClient.chat的history约定长期记忆 / 索引SQLite core/processor.pycore/rag.py入库RAGRAGClient.searchsearch_memoriesChatAgent._needs_rag工具core/tools.py技能skills/**/SKILL.mdgenerate_*.py规划core/planner.pyChatAgent._execute_plan建议自学者从main_agent.py进沿chat_agent.py→planner.py→skills/diary/scripts/generate.py→processor.py/rag.py走通一条用户路径再对照上表查漏补缺。下文附录再说明该仓库作为「日记应用」时前后端如何配合、如何启动——与 Agent 模块主线分开阅读即可。附录示例应用 MOSAIC 的前后端与使用方式前文把Agent 结构讲完了MOSAIC本身是这一结构的一种产品化输出本地日记/碎片记录 生成日记、想法、情绪洞察 对话。这里只交代系统怎么拼在一起、怎么用起来细节仍以仓库 README / README_CN 为准。后端单进程 FastAPI入口api/server.py创建FastAPI应用注册各路由模块并在启动时init_db()、后台启动Processor线程见前文「记忆索引」。HTTP APIapi/routes/下按资源拆分例如entries/write/generate/diaries/mood/chat——浏览器与脚本通过这些端点读写数据、触发生成、会话聊天。静态前端同一进程挂载frontend/到/static根路径/直接返回index.html因此无需单独前端构建流水线即可本地访问。本地模型核心业务通过core/llm.py调 Ollama无云端大模型 API 依赖。典型本地启动需已安装并运行 Ollamauvicorn api.server:app--reload--host127.0.0.1--port8000浏览器打开http://127.0.0.1:8000即可加载内置 UI。前端静态页面 Alpine.js位置frontend/index.html、app.js、style.css等。形态多 Tab如输入、手写、对话、日记/想法/情绪日历、时序长卷等通过fetch调用上文 API交互规则例如日历上有无数据、手写与生成内容的区分在 README_CN 中有说明。与 Agent 的关系前端不实现「规划/路由」只负责展示与触发命令Chat、生成类操作由后端路由到前文所述MainAgent/ 各 Agent 与技能脚本。Telegram 机器人可选第二条入口入口模块python -m bot.telegram_bot见仓库说明。作用与 Web 并行的一条消息入口用户发文本、命令同样进入MainAgent那一套解析与对话链路适合移动端随手记。配置.env中配置 Bot Token 等与纯 Web 使用前端的场景互不冲突。环境要点速览Python、Ollama及所选 chat / embedding 模型、可选Telegram Bot Token。依赖安装与模型拉取、.env示例等以仓库内pyproject.toml、.env.example与 README 为准。