1、布景取目标
2025 年 Agent（智能体）范围连续开展，Manus（本文屡次误写为 “minus”“manners”，为智能数据阐发标的目的的出名 Agent 框架）拉出时反应剧烈，早期聘请码密缺，盛开体会后果结果优良，成为浩瀚 Agent 框架的对于标工具。Manus 的中心劣势依靠强代码才气的狂言语模子（如Cloud 系列模子，测度为具备下阶代码天生取调试才气的年夜模子，本文拼写可以存留倾向），但是其民间已启源框架，仅能获得部门提醒词汇。
原文目标：鉴于LangGraph（LangChain 拉出的图构造智能体框架，颠末 “节面 + 边” 界说形状化、轮回式事情流），复现浅易版 Manus，散焦 “智能数据阐发陈述天生” 场景，完毕从需要输出到陈述输出的齐过程主动化。
2、中心道理：鉴于 LangGraph 的智能体架构
LangGraph 的中心是 “图构造”，颠末 “节面”（功用单位）战 “边”（节面间跳转逻辑）界说事情流，统统中心成果保存于 “形状” 中，保证过程可回溯、可迭代。原体系的架构设想完整环绕 “数据阐发陈述天生” 需要睁开，具体分为过程设想、形状办理、东西界说三年夜模块。
3、齐过程设想（节面取跳转逻辑）
原体系的事情流颠末 5 个中心节面完毕，过程关环以下：
开端节面 →方案 节面 → 施行节面 → 革新方案节面 → （轮回）→ 陈述天生节面 →完毕 节面
1. 节面功用取跳转划定规矩
节面称呼
|
中心功用
|
输出实质
|
输出实质
|
跳转逻辑
|
方案节面
|
按照用户需要创立施行方案
|
用户输出（如 “鉴于 Kaggle 师长教师数据天生阐发陈述”）、方案创立提醒词汇
|
契合格局的施行方案（露思考历程、目标、子方案）
|
天生方案后，自愿跳转至 “施行节面”
|
施行节面
|
挪用东西施行 “已完毕子方案”，曲至子方案完毕
|
目前施行方案、第一个已完毕子方案、施行提醒词汇、3 个中心东西
|
东西挪用记载（称呼、参数、成果）、子方案施行归纳
|
子方案完毕后，跳转至 “革新方案节面”；若子方案为 “天生陈述”，间接跳转至 “陈述天生节面”
|
革新方案节面
|
按照施行成果革新方案，仅改正已完毕子方案
|
目前方案、施行节面输出的归纳 /后果 、革新方案提醒词汇
|
革新后的施行方案（已经完毕子方案标识表记标帜为 “completed”，已完毕仍为 “pending”）
|
革新后跳转至 “施行节面”，鉴别可否另有已完毕子方案
|
陈述天生节面
|
鉴于统统中心成果天生终极阐发陈述
|
统统子方案归纳、东西挪用记载、陈述天生提醒词汇
|
终极阐发陈述（以文献方法输出，没有存进形状）
|
陈述天生后，跳转至 “完毕节面”
|
形状（隐性）
|
保存齐过程数据，供统统节面挪用
|
各节面输出的用户输出、方案、东西记载、归纳
|
各节面所需的汗青数据
|
无跳转，仅动作数据载体
|
2. 施行方案格局界说
方案节面天生的施行方案需严峻依照如下格局（保证后绝节面可剖析）：
salt：模子的思考历程（如 “用户需要阐发师长教师数据，需先减载数据、洗濯非常值、统计目标、画图，最初天生陈述”）；
goal：终极目标（如 “鉴于 Kaggle 师长教师习惯取偏见数据，天生包罗数据概略、洗濯历程、枢纽目标、可望化图表的阐发陈述”）；
steps（子方案列表）：每一个子方案露 3 个字段：
title：子方案扼要归纳综合（如 “减载 CSV 数据”）；
description：子方案具体分析（如 “使用 pandas 库读与数据文献，减载为 DataFrame 工具，检察数据前 5 止、形状及字段范例”）；
status：子方案形状（仅二种：pending（待施行）、completed（已经完毕））。
4、枢纽组件完毕
1. 提醒词汇工程（中心参照 Manus 民间提醒词汇）
提醒词汇是 Agent 决议计划的 “指令脚册”，原体系的提醒词汇分为 4 类，部门参照 Manus 保守的民间提醒词汇，保证取 Manus 逻辑对于齐：
提醒词汇范例
|
用处
|
特别分析
|
方案创立提醒词汇
|
辅导模子按照用户需要天生 “salt+goal+steps” 格局的施行方案
|
输出格局严峻束缚，制止后绝剖析报错
|
革新方案提醒词汇
|
辅导模子按照施行成果革新方案，仅改正已完毕子方案
|
输出格局取 “方案创立提醒词汇” 完整不合，低落节面剖析庞大度
|
施行节面提醒词汇
|
辅导模子鉴别 “挪用哪一个东西”“怎样施行子方案”，并正在完毕后输出归纳
|
大白东西挪用划定规矩（如 “代码报错需劣先用字符串交流东西改正，而非沉写”）
|
陈述天生提醒词汇
|
辅导模子鉴于中心成果（数据概略、洗濯记载、目标、图表）撰写华文陈述
|
自界说提醒词汇，散焦 “可读性”“逻辑性”，无参照 Manus（果民间陈述提醒词汇已保守）
|
2. 东西界说（3 个中心东西，笼盖数据阐发齐过程）
东西是 Agent 取内部情况接互的 “四肢举动”，原体系仅界说中心东西，制止功用冗余：
东西称呼
|
功用
|
使用场景举例
|
创立文献东西
|
天生 Python 代码文献（如data_load.py、data_clean.py）
|
需施行数据阐发操纵时，模子挪用该东西天生对于应的 Python 剧本
|
字符串交流东西
|
仅交流 Python 代码中的毛病片断（如语法毛病、函数参数毛病）
|
代码施行报错时，制止沉更生成残破代码，削减token（模子处置文原的根本单元）消耗
|
施行 shell 号令东西
|
施行 “创立文献东西” 天生的.py 文献，获得尺度输出（施行成果）或者尺度毛病（报错）
|
运行python data_load.py，获得数据形状（如 (1000,8)）或者 “模块已导进” 等报错疑息
|
3.形状 办理（鉴于 LangGraph 扩大）
LangGraph 的 “形状” 是齐过程数据的 “堆栈”，原体系鉴于 LangGraph根底 形状类扩大，新删枢纽字段，保证数据可回溯：
形状字段
|
保存实质
|
特别处置
|
user_input
|
用户本初需要（如 “阐发 Kaggle 师长教师数据，天生陈述”）
|
齐程稳定，供统统节面参照
|
plan
|
方案节面天生的 “salt+goal+steps” 格局数据
|
革新方案节面会笼盖该字段，仅保存最新方案
|
observations
|
东西挪用记载（东西称呼、参数、成果）+ 子方案归纳
|
仅保存 “终极准确施行” 的记载（如代码调试 3 次后胜利，仅存第 3 次成果），制止滋扰模子
|
final_report
|
终极陈述实质（原体系已使用）
|
果陈述间接天生文献（如analysis_report.md），故没有存进形状
|
4. 节面代码逻辑（中心功用完毕）
节面是图构造的 “功用单位”，需颠末代码界说每一个节面的输出、处置、输出逻辑，如下为枢纽节面的中心逻辑：
（1）帮助函数（预处置东西）
parse_json(output)：剖析模子输出的 JSON 格局方案 /后果 ，处置格局毛病（如缺得逗号、引号没有匹配）；
extract_answer(thought)：从模子的 “思考历程”（salt）中提炼有用答复（如来除冗余表述，保存中心决议计划）。
（2）方案节面（create_plan_node）
def create_plan_node(state): # 1. 获得输出：用户需要 +方案 创立提醒词汇 user_input = state["user_input"] system_prompt = PLAN_CREATION_PROMPT #方案 创立提醒词汇（参照Manus） prompt = f"用户需要：{user_input}\n{system_prompt}"
# 2. 挪用模子天生方案 model_output = llm.generate(prompt) # llm为减载的年夜模子（如Qwen-235B）
# 3. 剖析方案并革新形状 parsed_plan = parse_json(model_output) state["plan"] = parsed_plan
# 4. 跳转至施行节面 return {"next_node": "execute_plan_node", "state": state}
（3）施行节面（execute_plan_node）
def execute_plan_node(state): # 1. 获得目前方案，选择第一个已完毕子方案 plan = state["plan"] pending_steps = [step for step in plan["steps"] if step["status"] == "pending"] current_step = pending_steps[0] # 仅施行第一个已完毕子方案
# 2. 特别鉴别：若子方案为“天生陈述”，间接跳转至陈述节面 if "天生陈述" in current_step["title"]: return {"next_node": "generate_report_node", "state": state}
# 3. 准备输出：用户需要 +以后 子方案 + 施行提醒词汇 + 东西列表 user_input = state["user_input"] system_prompt = EXECUTION_PROMPT # 施行节面提醒词汇（参照Manus） tools = [create_file_tool, string_replace_tool, execute_shell_tool] # 3个中心东西 prompt = f"用户需要：{user_input}\n目前子方案：{current_step}\n{system_prompt}"
# 4. 轮回挪用东西，曲至子方案完毕 observations = [] while True: # 挪用模子鉴别可否需要持续挪用东西 model_response = llm.generate(prompt + f"\n已经挪用东西记载：{observations}") tool_call = extract_tool_call(model_response) # 提炼东西挪用疑息
if not tool_call: # 模子中断挪用东西，子方案完毕 # 提炼子方案归纳，革新形状 step_su妹妹ary = extract_answer(model_response) observations.append({"su妹妹ary": step_su妹妹ary}) state["observations"].extend(observations) break
# 施行东西并记载成果 tool_name = tool_call["name"] tool_args = tool_call["args"] tool_result = run_tool(tool_name, tool_args) # 施行东西 observations.append({"tool": tool_name, "args": tool_args, "result": tool_result})
# 5. 跳转至革新方案节面 return {"next_node": "update_plan_node", "state": state}
（4）革新方案节面（update_plan_node）
def update_plan_node(state): # 1. 获得目前方案、施行成果 plan = state["plan"] latest_observation = state["observations"][-1] # 最新施行成果（子方案归纳）
# 2. 准备输出：目前方案 + 施行成果 + 革新提醒词汇 system_prompt = PLAN_UPDATE_PROMPT # 革新方案提醒词汇（参照Manus） prompt = f"目前方案：{plan}\n最新施行成果：{latest_observation}\n{system_prompt}"
# 3. 挪用模子革新方案 updated_plan = parse_json(llm.generate(prompt))
# 4. 革新形状，跳转至施行节面 state["plan"] = updated_plan return {"next_node": "execute_plan_node", "state": state}
（5）陈述天生节面（generate_report_node）
def generate_report_node(state): # 1. 获得统统中心成果（东西记载、子方案归纳） observations = state["observations"] user_input = state["user_input"]
# 2. 准备输出：中心成果 + 陈述天生提醒词汇 system_prompt = REPORT_GENERATION_PROMPT # 自界说华文提醒词汇 prompt = f"用户需要：{user_input}\n数据阐发历程记载：{observations}\n{system_prompt}"
# 3. 挪用模子天生陈述，颠末“创立文献东西”输出 report_content = llm.generate(prompt) create_file_tool.run(filename="analysis_report.md", content=report_content)
# 4. 跳转至完毕节面 return {"next_node": "end_node", "state": state}
5、图构造建立取运行
1. 图构造界说（LangGraph 语法）
颠末 LangGraph 的Graph类界说节面取边，仅需隐式界说 “开端→方案”“陈述→完毕” 的边，其余节面跳转颠末代码内乱next_node指定：
from langgraph import Graph# 1. 初初化图，指定形状范例graph = Graph(State) # State为自界说扩大形状类# 2. 增加节面graph.add_node("create_plan_node", create_plan_node)graph.add_node("execute_plan_node", execute_plan_node)graph.add_node("update_plan_node", update_plan_node)graph.add_node("generate_report_node", generate_report_node)graph.add_node("end_node", lambda state: state) #完毕 节面（无逻辑）# 3. 界说边（仅隐式界说枢纽跳转）graph.add_edge("start", "create_plan_node") # 开端→方案节面graph.add_edge("generate_report_node", "end_node") # 陈述→完毕节面# 4. 编译图app = graph.compile()
2. 运行示例（鉴于 Kaggle 师长教师数据）
（1）尝试输出
用户需要：“鉴于 Kaggle 师长教师习惯取偏见数据散（student_data.csv），天生数据阐发陈述，需包罗数据概略、缺得值处置、枢纽目标统计（均匀分、尺度好）、可望化图表（散布曲圆图、饼图）”。
（2）运行过程取输出
方案节面：天生 5 身材方案（减载数据→缺得值处置→统计目标→画造图表→天生陈述）；
施行节面 1：挪用 “创立文献东西” 天生data_load.py，挪用 “施行 shell 号令东西” 运行，输出数据形状（如 (2000, 10)）、字段范例，标识表记标帜该子方案为 “completed”；
革新方案节面：革新方案，仅保存 4 个已完毕子方案；
施行节面 2：天生data_clean.py，运行后发明缺得值，挪用 “字符串交流东西” 改正代码（增加fillna()），从头施行胜利，标识表记标帜该子方案为 “completed”；
轮回迭代：重复 “施行→革新” 过程，曲至 “统计目标”“画造图表” 子方案完毕；
陈述天生节面：鉴于统统中心成果，天生analysis_report.md，包罗数据概略、洗濯步调、目标表格、图表拔出 链交。
（3）留神事变
模子挑选：尝试使用Qwen-235B 模子（通义千问 235B，本文 “万万 3 235B” 为心误 / 拼写倾向），代码才气一般，存留屡次报错后才改正的情况；
宁静性：原体系间接正在当地施行 shell 号令，理论使用需使用沙箱情况（断绝的 Python 情况），制止歹意代码进犯。
6、归纳取扩大标的目的
1. 复现易度归纳
浅易版（原文）：易度高，散焦简单 “数据阐发陈述” 场景，中心依靠 “提醒词汇 + LangGraph 节面逻辑”，无需庞大架构；
残破版 Manus：易度下，需撑持多场景任务，依靠三因素：①强代码才气模子（如 Cloud 系列）；②静态过程掌握（如子方案劣先级调解）；③完美的先后端接互（如用户上传文献、陈述正在线预览）。
2. 后绝劣化标的目的
模子升级：交流为更强代码才气的模子（如 Claude 3 Opus、GPT-4 Turbo），削减代码调试次数；
沙箱散成：交进 Docker 等沙箱情况，保证当地施行宁静性；
多格局输出：撑持天生 PDF、Excel 格局陈述，而非仅 MD 文献；
用户接互劣化：增加 “半途改正需要”功用 ，许可用户正在施行过程当中调解陈述偏重面。
7、布景进修质料（1）整体框架图
（2）相干真现代码1）graph.py：数据构造界说（形状办理中心）
功用定位：界说齐过程的「数据载体格式」，保证各节面间数据通报的不合性（对于应前文「形状办理」模块）。
from langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import MemorySaverfrom state import Statefrom nodes import ( report_node, execute_node, create_planner_node, update_planner_node)
def _build_base_graph(): """Build and return the base state graph with all nodes and edges.""" builder = StateGraph(State) builder.add_edge(START, "create_planner") builder.add_node("create_planner", create_planner_node) builder.add_node("update_planner", update_planner_node) builder.add_node("execute", execute_node) builder.add_node("report", report_node) builder.add_edge("report", END) return builder
def build_graph_with_memory(): """Build and return the agent workflow graph with memory.""" memory = MemorySaver() builder = _build_base_graph() return builder.compile(checkpointer=memory)
def build_graph(): """Build and return the agent workflow graph without memory.""" # build state graph builder = _build_base_graph() return builder.compile()
graph = build_graph()
inputs = {"user_message": "对于所给文档截至阐发，天生阐发陈述，文档路子为student_habits_performance.csv", "plan": None, "observations": [], "final_report": ""}
graph.invoke(inputs, {"recursion_limit":100})2）nodes.py：东西界说（Agent 的 “四肢举动”）
功用定位：完毕前文提到的 3 个中心东西（创立文献、字符串交流、施行 shell 号令），供施行节面 / 陈述节面挪用，完毕取内部情况的接互。
import jsonimport loggingfrom typing import Annotated, Literalfrom langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessagefrom langgraph.types import Co妹妹and, interruptfrom langchain_openai import ChatOpenAIfrom state import Statefrom prompts import *from tools import *
llm = ChatOpenAI(model="**", temperature=0.0, base_url='**', api_key='**')
logger = logging.getLogger(__name__)logger.setLevel(logging.INFO)hander = logging.StreamHandler()hander.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')hander.setFormatter(formatter)logger.addHandler(hander)
def extract_json(text): if '```json' not in text: return text text = text.split('```json')[1].split('```')[0].strip() return text
def extract_answer(text): if '</think>' in text: answer = text.split("</think>")[-1] return answer.strip()
return text
def create_planner_node(state: State): logger.info("奸淫在运行Create Planner node奸淫") messages = [SystemMessage(content=PLAN_SYSTEM_PROMPT), HumanMessage(content=PLAN_CREATE_PROMPT.format(user_message = state['user_message']))] response = llm.invoke(messages) response = response.model_dump_json(indent=4, exclude_none=True) response = json.loads(response) plan = json.loads(extract_json(extract_answer(response['content']))) state['messages'] += [AIMessage(content=json.dumps(plan, ensure_ascii=False))] return Co妹妹and(goto="execute", update={"plan": plan})
def update_planner_node(state: State): logger.info("奸淫在运行Update Planner node奸淫") plan = state['plan'] goal = plan['goal'] state['messages'].extend([SystemMessage(content=PLAN_SYSTEM_PROMPT), HumanMessage(content=UPDATE_PLAN_PROMPT.format(plan = plan, goal=goal))]) messages = state['messages'] while True: try: response = llm.invoke(messages) response = response.model_dump_json(indent=4, exclude_none=True) response = json.loads(response) plan = json.loads(extract_json(extract_answer(response['content']))) state['messages']+=[AIMessage(content=json.dumps(plan, ensure_ascii=False))] return Co妹妹and(goto="execute", update={"plan": plan}) except Exception as e: messages += [HumanMessage(content=f"json格局毛病:{e}")]
def execute_node(state: State): logger.info("奸淫在运行execute_node奸淫")
plan = state['plan'] steps = plan['steps'] current_step = None current_step_index = 0
# 获得第一个已完毕STEP for i, step in enumerate(steps): status = step['status'] if status == 'pending': current_step = step current_step_index = i break
logger.info(f"目前施行STEP:{current_step}")
## 此处不过简朴跳转到report节面，理论该当按照目前STEP的描绘截至鉴别 if current_step is None or current_step_index == len(steps)-1: return Co妹妹and(goto='report')
messages = state['observations'] + [SystemMessage(content=EXECUTE_SYSTEM_PROMPT), HumanMessage(content=EXECUTION_PROMPT.format(user_message=state['user_message'], step=current_step['description']))]
tool_result = None while True: response = llm.bind_tools([create_file, str_replace, shell_exec]).invoke(messages) response = response.model_dump_json(indent=4, exclude_none=True) response = json.loads(response) tools = {"create_file": create_file, "str_replace": str_replace, "shell_exec": shell_exec} if response['tool_calls']: for tool_call in response['tool_calls']: tool_name = tool_call['name'] tool_args = tool_call['args'] tool_result = tools[tool_name].invoke(tool_args) logger.info(f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}") messages += [ToolMessage(content=f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}", tool_call_id=tool_call['id'])]
elif '<tool_call>' in response['content']: tool_call = response['content'].split('<tool_call>')[-1].split('</tool_call>')[0].strip()
tool_call = json.loads(tool_call)
tool_name = tool_call['name'] tool_args = tool_call['args'] tool_result = tools[tool_name].invoke(tool_args) logger.info(f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}") messages += [ToolMessage(content=f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}", tool_call_id=tool_call['id'])] else: break
logger.info(f"目前STEP施行归纳:{extract_answer(response['content'])}")
state['messages'] += [AIMessage(content=extract_answer(response['content']))] if tool_result: state['observations'] += [ToolMessage(content=f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}", tool_call_id=tool_call['id'])] state['observations'] += [AIMessage(content=extract_answer(response['content']))]
return Co妹妹and(goto='update_planner', update={'plan': plan})
def report_node(state: State): """Report node that write a final report.""" logger.info("奸淫在运行report_node奸淫")
observations = state.get("observations") messages = observations + [SystemMessage(content=REPORT_SYSTEM_PROMPT)]
while True: response = llm.bind_tools([create_file, shell_exec]).invoke(messages) response = response.model_dump_json(indent=4, exclude_none=True) response = json.loads(response) tools = {"create_file": create_file, "shell_exec": shell_exec} if response['tool_calls']: for tool_call in response['tool_calls']: tool_name = tool_call['name'] tool_args = tool_call['args'] tool_result = tools[tool_name].invoke(tool_args) logger.info(f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}") messages += [ToolMessage(content=f"tool_name:{tool_name},tool_args:{tool_args}\ntool_result:{tool_result}", tool_call_id=tool_call['id'])] else: break
return {"final_report": response['content']}
3）prompts.py：提醒词汇工程（Agent 的 “指令脚册”）
功用定位：保存前文提到的 4 类中心提醒词汇，辅导 LLM 天生方案、革新方案、挪用东西、天生陈述（部门参照 Manus 民间提醒词汇）。
PLAN_SYSTEM_PROMPT = f"""You are an intelligent agent with autonomous planning capabilities, capable of generating detailed and executable plans based on task objectives.
<language_settings>- Default working language: **Chinese**- Use the language specified by user in messages as the working language when explicitly provided- All thinking and responses must be in the working language</language_settings>
<execute_environment>System Information- Base Environment: Python 3.11 + Ubuntu Linux (minimal version)- Installed Libraries: pandas, openpyxl, numpy, scipy, matplotlib, seaborn
Operational Capabilities1 File Operations- Create, read, modify, and delete files- Organize files into directories/folders- Convert between different file formats2 Data Processing- Parse structured data (XLSX, CSV, XML)- Cleanse and transform datasets- Perform data analysis using Python libraries- Chinese font file path: SimSun.ttf </execute_environment>"""
PLAN_CREATE_PROMPT = '''You are now creating a plan. Based on the user's message, you need to generate the plan's goal and provide steps for the executor to follow.
Return format requirements are as follows:- Return in JSON format, must comply with JSON standards, cannot include any content not in JSON standard- JSON fields are as follows: - thought: string, required, response to user's message and thinking about the task, as detailed as possible - steps: array, each step contains title and description - title: string, required, step title - description: string, required, step description - status: string, required, step status, can be pending or completed - goal: string, plan goal generated based on the context- If the task is determined to be unfeasible, return an empty array for steps and empty string for goal
EXAMPLE JSON OUTPUT:{{ "thought": "" "goal": "", "steps": [ {{ "title": "", "description": "" "status": "pending" }} ],}}
Create a plan according to the following requirements:- Provide as much detail as possible for each step- Break down complex steps into multiple sub-steps- If multiple charts need to be drawn, draw them step by step, generating only one chart per step
User message:{user_message}/no_think'''
UPDATE_PLAN_PROMPT = """You are updating the plan, you need to update the plan based on the context result.- Base on the lastest content delete, add or modify the plan steps, but don't change the plan goal- Don't change the description if the change is small- Status: pending or completed- Only re-plan the following uncompleted steps, don't change the completed steps- Keep the output format consistent with the input plan's format.
Input:- plan: the plan steps with json to update- goal: the goal of the plan
Output:- the updated plan in json format
Plan:{plan}
Goal:{goal}/no_think"""
EXECUTE_SYSTEM_PROMPT = """You are an AI agent with autonomous capabilities.
<intro>You excel at the following tasks:1. Data processing, analysis, and visualization2. Writing multi-chapter articles and in-depth research reports3. Using progra妹妹ing to solve various problems beyond development</intro>
<language_settings>- Default working language: **Chinese**- Use the language specified by user in messages as the working language when explicitly provided- All thinking and responses must be in the working language</language_settings>
<system_capability>- Access a Linux sandbox environment with internet connection- Write and run code in Python and various progra妹妹ing languages- Utilize various tools to complete user-assigned tasks step by step</system_capability>
<event_stream>You will be provided with a chronological event stream (may be truncated or partially omitted) containing the following types of events:1. Message: Messages input by actual users2. Action: Tool use (function calling) actions3. Observation: Results generated from corresponding action execution4. Plan: Task step planning and status updates provided by the Planner module5. Other miscellaneous events generated during system operation</event_stream>
<agent_loop>You are operating in an agent loop, iteratively completing tasks through these steps:1. Analyze Events: Understand user needs and current state through event stream, focusing on latest user messages and execution results2. Select Tools: Choose next tool call based on current state, task planning3. Iterate: Choose only one tool call per iteration, patiently repeat above steps until task completion</agent_loop>
<file_rules>- Use file tools for reading, writing, appending, and editing to avoid string escape issues in shell co妹妹ands- Actively save intermediate results and store different types of reference information in separate files- When merging text files, must use append mode of file writing tool to concatenate content to target file- Strictly follow requirements in <writing_rules>, and avoid using list formats in any files except todo.md</file_rules>
<coding_rules>- Must save code to files before execution; direct code input to interpreter co妹妹ands is forbidden- Write Python code for complex mathematical calculations and analysis</coding_rules>
<writing_rules>- Write content in continuous paragraphs using varied sentence lengths for engaging prose; avoid list formatting- Use prose and paragraphs by default; only employ lists when explicitly requested by users- All writing must be highly detailed with a minimum length of several thousand words, unless user explicitly specifies length or format requirements- When writing based on references, actively cite original text with sources and provide a reference list with URLs at the end- For lengthy documents, first save each section as separate draft files, then append them sequentially to create the final document- During final compilation, no content should be reduced or su妹妹arized; the final length must exceed the sum of all individual draft files</writing_rules>"""
EXECUTION_PROMPT = """<task>Select the most appropriate tool based on <user_message> and context to complete the <current_step>.</task>
<requirements>1. Must use Python for data processing and chart generation2. Charts default to TOP10 data unless otherwise specified3. Su妹妹arize results after completing <current_step> (Su妹妹arize only <current_step>, no additional content should be generated.)</requirements>
<additional_rules>1. Data Processing: - Prioritize pandas for data operations - TOP10 filtering must specify sort criteria in co妹妹ents - No custom data fields are allowed2. Code Requirements: - Must use the specified font for plotting. Font path: *SimSun.ttf* - The chart file name must reflect its actual content. - Must use *print* statements to display intermediate processes and results.</additional_rules>
<user_message>{user_message}</user_message>
<current_step>{step}</current_step>"""
REPORT_SYSTEM_PROMPT = """<goal>您是陈述天生大师，您需要按照已经有的高低文疑息（数据疑息、图表疑息等），天生一份有代价的陈述。</goal>
<style_guide>- 使用表格战图表展示数据- 没有要描绘图表的局部数据，只描绘具备清楚意思的目标- 天生丰硕有代价的实质，从多个维度分离，制止过于简单</style_guide>
<attention>- 陈述契合数据阐发陈述格局，包罗但是没有限于阐发布景，数据概括，数据开掘取可望化，阐发倡议取论断等（可按照理论情况截至扩大）- 可望化图表必需拔出 阐发历程，没有患上零丁展示或者以附件方法列出- 陈述中没有患上出现代码施行毛病相干疑息- 起首天生各身材陈述，而后兼并统统子陈述文献获得残破陈述- 以文献方法展示阐发陈述</attention>"""4）state.py：中心节面完毕（Agent 的 “年夜脑”）
功用定位：完毕前文提到的 4 个中心节面（方案节面、施行节面、革新方案节面、陈述节面），是全部体系的逻辑中心。
from langgraph.graph import MessagesStatefrom typing import Optional, List, Dict, Literal
from enum import Enumfrom typing import List, Optional
from pydantic import BaseModel, Field
class Step(BaseModel): title: str = "" description: str = "" status: Literal["pending", "completed"] = "pending"
class Plan(BaseModel): goal: str = "" thought: str = "" steps: List[Step] = []
class State(MessagesState): user_message: str = "" plan: Plan observations: List = [] final_report: str = ""
5）tools.py：图构造建立取事情流触收
功用定位：用 LangGraph 调整统统节面，建立 “开端→方案→施行→革新→陈述→完毕” 的关环，并触收事情流。
from langchain_core.tools import toolimport osimport tracebackimport subprocess@tooldef create_file(file_name, file_contents): """ Create a new file with the provided contents at a given path in the workspace.
args: file_name (str): Name to the file to be created file_contents (str): The content to write to the file """ try:
file_path = os.path.join(os.getcwd(), file_name) os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w') as file: file.write(file_contents)
return { "message": f"Successfully created file at {file_path}" }
except Exception as e: return { "error": str(e) }
@tooldef str_replace(file_name, old_str, new_str): """ WordStr specific text in a file.
args: file_name (str): Name to the target file old_str (str): Text to be replaced (must appear exactly once) new_str (str): WordStrment text """ try: file_path = os.path.join(os.getcwd(), file_name) with open(file_path, "r") as file: content = file.read()
new_content = content.replace(old_str, new_str, 1)
with open(file_path, "w") as file: file.write(new_content)
return {"message": "Successfully replaced '{old_str}' with '{new_str}' in {file_path}"} except Exception as e: return {"error": f"Error replacing '{old_str}' with '{new_str}' in {file_path}: {str(e)}"}
@tooldef send_message(message: str): """ send a message to the user
args: message: the message to send to the user """
return message
@tooldef shell_exec(co妹妹and: str) -> dict: """ 正在指定的 shell 会话中施行号令。
参数: co妹妹and (str): 要施行的 shell 号令
前去: dict:包括 如下字段： - stdout: 号令的尺度输出 - stderr: 号令的尺度毛病 """
try: # 施行号令 result = subprocess.run( co妹妹and, shell=True, cwd=os.getcwd(), capture_output=True, text=True, check=False )
# 前去成果 return {"message":{"stdout": result.stdout,"stderr": result.stderr}}
except Exception as e: return {"error":{"stderr": str(e)}} |
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
照妖镜
广告
