Agent AI 工作流：从概念到实战的基础知识指南

1. 什么是 Agent AI 工作流Agent AI 工作流是指以大型语言模型LLM为核心通过编排多个智能体Agent或工具调用完成复杂任务的系统化流程。与传统单次问答不同Agent 工作流强调规划、执行、反馈与迭代让模型能够自主拆解目标、调用外部工具、读取结果并调整策略最终交付可靠成果。一个典型的 Agent 工作流包含以下核心要素任务目标用户以自然语言描述需求系统将其转化为可执行的结构化指令。规划器Planner将复杂任务拆解为多个子步骤并决定执行顺序。执行器Executor调用代码、API、数据库或其它工具完成具体操作。反馈回路Feedback Loop根据执行结果判断是否达成目标必要时重新规划或修正。记忆与上下文在多次交互中保留关键信息避免重复提问或丢失状态。这种设计让 AI 从「被动回答」升级为「主动完成任务」适用于数据分析、自动化运维、客户服务、代码生成等场景。2. Agent 工作流的核心设计模式在实际工程中Agent 工作流通常采用以下几种设计模式2.1 单 Agent 工具调用单个 LLM 实例通过函数调用Function Calling / Tool Calling使用外部工具。模型根据用户输入决定调用哪个工具、传入什么参数再根据工具返回结果生成最终回复。这是最基础的模式适合任务边界清晰、工具数量有限的场景。2.2 多 Agent 协作多个 Agent 各司其职通过消息传递或共享状态协同工作。例如一个 Agent 负责理解用户意图另一个负责查询数据库第三个负责生成报告。多 Agent 模式适合复杂任务但需要设计好通信协议和任务分配策略。2.3 规划-执行-反思循环Agent 先制定计划逐步执行并在每步之后评估结果。如果发现偏差或错误则重新规划。这种「反思」机制能显著提升任务成功率尤其适合需要多步推理或外部反馈的场景。2.4 人机协同Human-in-the-Loop在关键节点引入人工确认或干预例如高风险操作、模糊指令或超出模型能力范围的决策。人机协同能兼顾自动化效率与安全性。3. 关键技术组件构建 Agent 工作流通常需要以下技术组件组件作用常见实现LLM 引擎理解意图、生成代码或文本GPT、Claude、开源模型工具调用层让模型调用外部函数或 APIFunction Calling、Tool Use代码执行沙箱安全运行模型生成的代码Docker、受限 exec 环境数据存储保存状态、记忆与中间结果数据库、向量库、文件系统编排框架管理 Agent 生命周期与任务流转LangChain、自研 Pipeline其中代码执行沙箱是数据类 Agent 的关键。模型生成的 Python 代码需要在受控环境中运行既要保证功能完整又要防止恶意操作或资源滥用。4. 一个实战示例一个眼镜店的客服 Agent下面通过一个库存管理场景展示 Agent 工作流的完整链路。用户用自然语言提问系统自动生成 TinyDB 查询代码并执行最终返回友好回复。4.1 整体流程用户自然语言输入 │ ▼ generate_llm_code() ┌─────────────────────────────────┐ │ PROMPT schema 规则 问题 │ │ LLM 生成 TinyDB Python 代码 │ │ 包裹在 execute_python 标签里 │ └─────────────────────────────────┘ │ ▼ extract_execute_code() ┌─────────────────────────────────┐ │ 用 re 提取标签内的代码 │ │ 无标签则直接用原始文本 │ └─────────────────────────────────┘ │ ▼ execute_generated_code() ┌─────────────────────────────────┐ │ exec(code, SAFE_GLOBALS, │ │ SAFE_LOCALS) │ │ 捕获 stdout / error │ │ 提取 answer_text / STATUS │ └─────────────────────────────────┘ │ ▼ 返回 answer_text 给用户4.2 提示词设计提示词是 Agent 工作流的大脑。它需要明确告诉模型数据库结构是什么、有哪些规则、如何输出。以下是一个精简示例PROMPT You are a senior data assistant. PLAN BY WRITING PYTHON CODE USING TINYDB. Database Schema Samples (read-only): {schema_block} Execution Environment (already imported/provided): Variables: db, inventory_tbl, transactions_tbl Helpers: get_current_balance(tbl), next_transaction_id(tbl) PLANNING RULES: Extract ALL filters from the user_request. Build TinyDB queries dynamically with Query(). If intent is ambiguous, do read-only (DRY RUN). OUTPUT CONTRACT: Return ONLY executable Python between these tags: execute_python your python /execute_python User request: {question} 4.3 代码生成与执行模型返回的代码通过正则提取再放入受限的命名空间执行。执行环境只暴露必要的变量和函数避免模型访问无关资源4.3.1 代码生成这里使用的是OpenAI的模型,首先要从.env文件中加载API_KEY然后实例化一个OpenAI client。# 懒加载避免 import 时因缺少 API Key 直接崩溃 client: OpenAI | None None def get_client(): global client if client is None: api_key os.getenv(OpenAI_API_KEY) if not api_key: raise EnvironmentError(OpenAI_API_KEY not set) client OpenAI(api_keyapi_key) return client def generate_llm_code( user_question: str, inventory_tbl, transactions_tbl, model:str MODEL, temperature: float 0.2, ): schema_block inv_utils.build_schema_block(inventory_tbl, transactions_tbl) full_prompt PROMPT.format(schema_blockschema_block, questionuser_question) client get_client() resp client.chat.completions.create( modelmodel, temperaturetemperature, messages[ { role: system, content: You write safe, well-commented TinyDB code to handle data questions and updates. }, {role: user, content: full_prompt}, ], ) content resp.choices[0].message.content or return content4.3.2 提取生成的代码由于提示词中写到代码是生成在标签execute_python /execute_python之间的所以用正则表达式来从content中提取生成的代码。过程比较简单有标签的提取标签之间的内容没有标签的直接返回text这是因为大模型生成代码是有随机性的有时候会没有标签。# --- Helper: extract code between execute_python.../execute_python --- def extract_execute_code(text:str)-str: Returns the Python code inside execute_python.../execute_python. If no tags are found, assumes text is already raw Python code. :param text: :return: if not text: raise RuntimeError(Empty content passed to code executor) m re.search(rexecute_python(.*?)/execute_python,text,re.DOTALL | re.IGNORECASE) if m : # print(f\n{ * 30}extract_execute_code{ * 30}) # print(m.group(1).strip()) return m.group(1).strip() else: # print(f\n{ * 30}no code to extract{ * 30}) # print(text.strip()) text.strip() return text4.3.3 执行生成的代码代码执行是比较重要的一个环节要保证在sandbox沙盒环境中执行不能影响到当前系统。┌─────────────────────────────────────────────────────────────┐ │ execute_generated_code() │ │ 入参: code_or_content, db, inventory_tbl, │ │ transactions_tbl, user_request │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ extract_execute_code(code_or_content) │ │ │ │ 有 execute_python 标签? │ │ ├── YES → 提取标签内代码 .strip() │ │ └── NO → 直接用原始文本 .strip() │ └─────────────────────────┬───────────────────────────────────┘ │ code (纯 Python 字符串) ▼ ┌─────────────────────────────────────────────────────────────┐ │ 构建执行环境 │ │ │ │ SAFE_BUILTINS { print, len, min, __import__, ... } │ │ │ │ SAFE_GLOBALS { __builtins__: SAFE_BUILTINS, │ │ Query, get_current_balance, │ │ next_transaction_id, user_request } │ │ │ │ SAFE_LOCALS { db, inventory_tbl, transactions_tbl } │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 启动子线程执行代码 (daemonTrue) │ │ │ │ Thread._run(): │ │ redirect_stdout(stdout_buffer) │ │ └── exec(code, SAFE_GLOBALS, SAFE_LOCALS) │ │ ├── 成功 → SAFE_LOCALS 中也用来保存执行代码过程中 │ 产生的结果变量新增变量 │ │ │ (answer_text / STATUS / results ...) │ │ └── 异常 → err_text traceback │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ t.join(timeout10s) │ │ 避免生成的代码中有死循环 │ │ 10秒内完成? │ │ ├── YES → 继续 │ │ └── NO → timed_outTrue │ │ err_textCode execution timed out │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 收集结果 │ │ │ │ printed stdout_buffer.getvalue() ← print 输出 │ │ │ │ answer SAFE_LOCALS.get(answer_text) │ │ or SAFE_LOCALS.get(answer_rows) │ │ or SAFE_LOCALS.get(answer_json) │ └─────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 返回 dict │ │ │ │ { │ │ code : 实际执行的代码, │ │ stdout : print 输出内容, │ │ error : 异常信息 / None, │ │ timed_out : True / False, │ │ answer : 给用户的回复文本, │ │ transactions_tbl: 执行后的交易表快照, │ │ inventory_tbl : 执行后的库存表快照 │ │ } │ └─────────────────────────────────────────────────────────────┘代码如下def execute_generated_code( code_or_content: str, db, inventory_tbl, transactions_tbl, user_request: Optional[str] None, ) - Dict[str, Any]: Execute code in a controlled namespace. Accepts either raw Python code OR full content with execute_python tags. Returns minimal artifacts: stdout, error, and extracted answer. code extract_execute_code(code_or_content) # 只开放必要的内置函数防止 open/eval 等危险操作 SAFE_BUILTINS { print: print, len: len, range: range, min: min, max: max, sum: sum, str: str, int: int, float: float, list: list, dict: dict, bool: bool, enumerate: enumerate, zip: zip, sorted: sorted, isinstance: isinstance, hasattr: hasattr, True: True, False: False, None: None, __import__: __import__, } SAFE_GLOBALS { __builtins__: SAFE_BUILTINS, Query: Query, get_current_balance: inv_utils.get_current_balance, next_transaction_id: inv_utils.next_transaction_id, user_request: user_request or , } SAFE_LOCALS { db: db, inventory_tbl: inventory_tbl, transactions_tbl: transactions_tbl, } err_text None timed_out False stdout_buffer io.StringIO() # 用 redirect_stdout 替换 sys.stdout 直接赋値线程安全 # 加超时保护防止 LLM 生成的代码死循环卡住进程 def _run(): nonlocal err_text try: from contextlib import redirect_stdout with redirect_stdout(stdout_buffer): exec(code, SAFE_GLOBALS, SAFE_LOCALS) except Exception: err_text traceback.format_exc() t threading.Thread(target_run, daemonTrue) t.start() t.join(timeout10) # 最多等待 10 秒 if t.is_alive(): timed_out True err_text Code execution timed out ( 10s) printed stdout_buffer.getvalue().strip() answer ( SAFE_LOCALS.get(answer_text) or SAFE_LOCALS.get(answer_rows) or SAFE_LOCALS.get(answer_json) ) return { code: code, stdout: printed, error: err_text, timed_out: timed_out, answer: answer, transactions_tbl: transactions_tbl.all(), inventory_tbl: inventory_tbl.all(), }需要注意以下几点:__builtins__ 需要设置访问范围避免LLM调用权限较高的危险函数比如open/eval等用contextlib.redirect_stdout重定向输出避免线程不安全exec()加超时保护避免生成代码中有死循环最后是测试阶段提示词是“退还两个飞行眼镜”prompt_aviator Return 2 Aviator sunglasses I bought last week. # Generate the plan-as-code (FULL content; may include execute_python tags) full_content_aviator generate_llm_code( prompt_aviator, inventory_tbl, transactions_tbl, modelMODEL, temperature0.2, ) # Execute the generated plan for the round-sunglasses question result execute_generated_code( full_content_aviator, # the full LLM response you generated earlier dbdb, inventory_tblinventory_tbl, transactions_tbltransactions_tbl, user_requestprompt_aviator, # e.g., Return 2 aviator sunglasses I bought last week. ) # Peek at exactly what Python the plan executed print(result[answer])生成的代码为execute_python # Parsed intent: { # action: return, # item_name: Aviator, # quantity: 2, # price_range: None, # keyword: None # } # The request is a RETURN (refund) of 2 Aviator sunglasses. # This is a state-changing action - ACTIONmutate, SHOULD_MUTATETrue. from tinydb import Query from datetime import datetime Item Query() ITEM_NAME Aviator QTY 2 ACTION mutate SHOULD_MUTATE True # 1) Look up the requested item by exact style name (case-insensitive match). matches inventory_tbl.search( Item.name.test(lambda v, nmITEM_NAME: (v or ).strip().lower() nm.lower()) ) if not matches: # No such item in catalog - cannot process return, suggest closest alternative. STATUS no_match # Try to find a similarly named item as a helpful alternative (simple substring check). alt inventory_tbl.search( Item.name.test(lambda v, nmITEM_NAME: nm.lower()[:4] in (v or ).lower()) ) if alt: alt_name alt[0][name] answer_text fWe couldnt find {ITEM_NAME} in our records, but we do carry {alt_name}—would that be the item you meant to return? else: answer_text fWe couldnt find {ITEM_NAME} in our inventory, so we cant process that return. Could you confirm the item name? print(f[LOG] ACTION{ACTION} STATUS{STATUS} No matching item found for return of {ITEM_NAME}.) else: # 2) Item found - proceed with return (no stock-availability check needed since were adding stock back). item matches[0] unit_price item[price] line_total unit_price * QTY # 3) Update inventory: increase quantity_in_stock by QTY. new_qty item[quantity_in_stock] QTY inventory_tbl.update({quantity_in_stock: new_qty}, Item.item_id item[item_id]) # 4) Record the refund transaction: money flows OUT of the register (negative amount). current_balance get_current_balance(transactions_tbl) refund_amount -line_total new_balance current_balance refund_amount txn_id next_transaction_id(transactions_tbl, prefixTXN) transactions_tbl.insert({ transaction_id: txn_id, customer_name: CUSTOMER_RETURN, transaction_summary: fReturn of {QTY} x {ITEM_NAME} (item_id{item[item_id]}), transaction_amount: refund_amount, balance_after_transaction: new_balance, timestamp: datetime.now().isoformat() }) STATUS success answer_text ( fYour return of {QTY} pairs of {ITEM_NAME} sunglasses has been processed fand a refund of ${line_total:.2f} has been issued. ) print(f[LOG] ACTION{ACTION} STATUS{STATUS} Returned {QTY}x {ITEM_NAME} f(item_id{item[item_id]}), refund${line_total:.2f}, fnew_stock{new_qty}, new_balance{new_balance:.2f}, txn{txn_id}) /execute_python执行结果Your return of 2 pairs of Aviator sunglasses has been processed and a refund of $160.00 has been issued.5.总结​​​​当前项目只调用了一次LLM在提示词里把代码最后输出的结果回答直接生成好了不需要再调用一次LLM来输出结果。answer_text ( fYour return of {QTY} pairs of {ITEM_NAME} sunglasses has been processed fand a refund of ${line_total:.2f} has been issued. )标准流程应该是ReAct,后面可以改进。第一轮 LLM └── 只生成查询代码inventory_tbl.search(...) exec() 执行 └── 只查询数据库得到原始数据 results[{...}] 第二轮 LLM └── 拿到 results生成自然语言回复 Yes, we have Classic sunglasses for $60 输出 → 用户OpenAI-compatible 的resp.choices[0].message.content 取文本的结构ChatCompletionResponse ├── id # 请求唯一 ID ├── object # 固定值 chat.completion ├── created # 时间戳 ├── model # 实际使用的模型名 ├── usage # token 用量 │ ├── prompt_tokens │ ├── completion_tokens │ └── total_tokens └── choices: list # 通常只有 1 个元素 └── choices[0] ├── index # 0 ├── finish_reason # FinishReason.stop / .length / .tool_calls └── message ├── role # assistant ├── content # str ← 这就是 LLM 的回复文本 ├── tool_calls # 有 tool calling 时才有值 ├── tool_call_id ├── name └── reasoning_content # 思维链内容部分模型支持完整源码可在附件中下载。