各位访问大哥能否留一些宝贵建议给小弟没人的话就当个人日常blog我结合前面学过的知识给agent补充了一些功能流式输出、历史会话保存和读取、滑窗裁剪并仿照之前做过的网页版聊天机器人利用Streamlit给agent包装了一个web。因为本项目架构的各个功能模块都是参照之前做过的demo下面就仅做简单介绍不深入解释。第一部分是初始化内容包括接入模型api和接口网址、加载 Embedding 模型将文本转为向量、加载 token 计数器计算会话消耗的token、设置安全阈值token超出就触发滑窗裁剪、设置文件保存路径和连接 ChromaDB 向量数据库。root Path(__file__).parent.parent load_dotenv(root / .env) client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), ) emb_model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) encoder tiktoken.get_encoding(cl100k_base) MAX_TOKENS 100_000 SAVE_DIR Path(__file__).parent / agent_output HISTORY_FILE Path(__file__).parent / chat_history.json # ChromaDB chroma_client chromadb.PersistentClient(pathstr(Path(__file__).parent / chroma_db)) collection chroma_client.get_or_create_collection(nameknowledge_base)第二部分是工具编写包括工具函数定义、工具描述TOOLS和系统提示词system prompt。def load_knowledge(): if collection.count() 0: return know_file Path(__file__).parent / python_knowledge.txt with open(know_file, r, encodingutf-8) as f: text f.read() chunks [s.strip() 。 for s in text.split(。) if s.strip()] embeddings emb_model.encode(chunks).tolist() collection.add( ids[fchunk_{i} for i in range(len(chunks))], embeddingsembeddings, documentschunks, ) def search_knowledge(query: str) - str: results collection.query( query_embeddingsemb_model.encode([query]).tolist(), n_results3 ) if not results[documents][0]: return 未找到相关内容 return \n.join([f- {doc} for doc in results[documents][0]]) def calculator(expression: str) - str: allowed set(0123456789-*/.() ) if not all(c in allowed for c in expression): return f错误表达式包含不允许的字符 try: return str(eval(expression)) except Exception as e: return f计算错误{e} def save_file(filename: str, content: str) - str: SAVE_DIR.mkdir(exist_okTrue) filepath SAVE_DIR / filename with open(filepath, w, encodingutf-8) as f: f.write(content) return f文件已保存{filepath} TOOLS [ { type: function, function: { name: search_knowledge, description: 在 Python 知识库中搜索相关内容。当用户问 Python 相关的技术问题时使用, parameters: { type: object, properties: { query: {type: string, description: 搜索关键词或问题} }, required: [query] } } }, { type: function, function: { name: calculator, description: 执行数学计算。当需要做算术运算时使用, parameters: { type: object, properties: { expression: {type: string, description: 数学表达式如 35*2} }, required: [expression] } } }, { type: function, function: { name: save_file, description: 将内容保存为文件。当用户要求保存、导出、记录内容时使用, parameters: { type: object, properties: { filename: {type: string, description: 文件名如 summary.txt}, content: {type: string, description: 要保存的内容} }, required: [filename, content] } } } ] SYSTEM_PROMPT 你是小码一个智能助手。你可以使用工具完成任务 - search_knowledge: 搜索 Python 知识库 - calculator: 执行数学计算 - save_file: 保存内容为文件 工作方式逐步调用工具每步观察结果完成后向用户汇报。 规则用中文回答简洁清晰。任务完成后输出总结不要反复调同一个工具。第三部分是会话处理包含读取历史会话、保存会话内容、计算会话总消耗token和滑窗裁剪。def load_history() - list: if HISTORY_FILE.exists(): with open(HISTORY_FILE, r, encodingutf-8) as f: return json.load(f) return [] def save_history(history: list): with open(HISTORY_FILE, w, encodingutf-8) as f: json.dump(history, f, ensure_asciiFalse, indent2) def count_tokens(messages: list) - int: total 0 for msg in messages: total len(encoder.encode(msg.get(content, ) or )) return total def trim_history(history: list) - list: while len(history) 2 and count_tokens(history) MAX_TOKENS: history history[2:] return history第四部分是基于Function Calling的react循环是agent最核心的部分同时实现流式输出功能。def run_agent(user_query: str, history: list, status_container, placeholderNone) - str: Agent 主循环。placeholder 用于流式展示最终回答。 messages [ {role: system, content: SYSTEM_PROMPT}, ] history [ {role: user, content: user_query}, ] step 0 while step 10: step 1 response client.chat.completions.create( modeldeepseek-v3.2, messagesmessages, toolsTOOLS, tool_choiceauto, ) msg response.choices[0].message if msg.tool_calls: for tool_call in msg.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) status_container.write(f 第 {step} 步调用 {func_name}({func_args})) if func_name search_knowledge: result search_knowledge(**func_args) elif func_name calculator: result calculator(**func_args) elif func_name save_file: result save_file(**func_args) else: result f未知工具: {func_name} messages.append({ role: assistant, content: None, tool_calls: [tool_call] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) else: # 流式输出 if placeholder: stream client.chat.completions.create( modeldeepseek-v3.2, messagesmessages, streamTrue, ) full for chunk in stream: content chunk.choices[0].delta.content if content: full content placeholder.markdown(full) # 逐字刷新 return full return msg.content or 任务已完成 return 任务未在限定步数内完成请简化任务重试。第五部分是通过Streamlit实现界面化。st.set_page_config(page_title小码 Agent, page_icon, layoutwide) st.title( 小码 Agent) st.caption(多工具自主智能助手 — 查知识库 · 做计算 · 存文件) # 初始化知识库 load_knowledge() # 初始化会话状态 if conversation_history not in st.session_state: st.session_state.conversation_history load_history() history st.session_state.conversation_history # 侧边栏 with st.sidebar: st.header( 状态) st.metric(知识库段落, collection.count()) st.metric(对话轮数, len(history) // 2) token_count count_tokens(history) st.metric(Token 用量, f{token_count} / {MAX_TOKENS}) st.metric(保存的文件, len(list(SAVE_DIR.glob(*))) if SAVE_DIR.exists() else 0) st.divider() if st.button(️ 清空对话记忆): st.session_state.conversation_history [] save_history([]) st.rerun() # 显示历史消息 for msg in history: with st.chat_message(msg[role]): st.markdown(msg[content]) # 用户输入 if user_input : st.chat_input(输入你的任务...): # 显示用户消息 with st.chat_message(user): st.markdown(user_input)外部知识库内容如下Python 是一门解释型、面向对象的高级编程语言。 Python 由 Guido van Rossum 在 1989 年圣诞节期间开始设计1991 年首次发布。 Python 的名字来源于 Guido 喜爱的 BBC 电视节目《蒙提·派森的飞行马戏团》。 Python 的设计哲学强调代码可读性使用缩进来表示代码块而不是花括号。 Python 2.0 于 2000 年发布引入了列表推导式、垃圾回收机制。 Python 3.0 于 2008 年发布是一次不向后兼容的大版本更新。 Python 3.0 的主要变化包括print 变为函数、整数除法返回浮点数、Unicode 成为默认字符串类型。 Python 2.7 是 Python 2 系列的最后一个版本官方支持于 2020 年 1 月 1 日终止。 pip 是 Python 的包管理工具用于安装和管理第三方库。 Python 的标准库非常丰富涵盖文件操作、网络通信、正则表达式、JSON 处理等。 Python 最流行的 Web 框架包括 Django、Flask 和 FastAPI。 Python 在数据科学领域的主要库包括 NumPy、Pandas 和 Matplotlib。 Python 在机器学习领域的主要库包括 Scikit-learn、TensorFlow 和 PyTorch。 Python 的变量不需要声明类型是动态类型语言。 Python 支持多种编程范式包括面向对象、函数式和过程式编程。 Python 使用 GIL全局解释器锁来保证线程安全但这限制了多线程的并行性能。 Python 3.12 引入了子解释器为绕过 GIL 提供了新的可能性。测试网页版agent功能RAG向量检索测试网页版agent功能数字计算测试网页版agent功能内容保存测试询问知识库外的内容测试上下文关联以上测试结果均通过。总结今天给agent包装了一个网页因为网页端比终端界面更美观且更方便使用同时补充了流式输出、历史会话保存和读取、滑窗裁剪等功能。做到这里算是完成一次完整的agent应用开发只是功能还不是那么强大技术也算不上成熟。后面我会进一步学习来完善争取做个满意的版本上传github。第九天就到这吧。
