CodeAct作為AI輔助系統(tǒng)的一種先進(jìn)范式，實(shí)現(xiàn)了自然語言處理與代碼執(zhí)行能力的深度融合。通過構(gòu)建自定義代碼執(zhí)行代理，開發(fā)者能夠精確控制應(yīng)用程序中代碼的生成、執(zhí)行及管理流程。本文將詳細(xì)闡述如何利用LlamaIndex框架從底層構(gòu)建CodeAct Agent，深入剖析其內(nèi)部工作機(jī)制，以及如何在預(yù)構(gòu)建解決方案的基礎(chǔ)上進(jìn)行定制化擴(kuò)展。

CodeAct技術(shù)范式的核心特性

從技術(shù)架構(gòu)角度而言，CodeAct賦予AI助手以下關(guān)鍵能力：

1. 基于自然語言指令的代碼生成：將用戶語義需求轉(zhuǎn)換為可執(zhí)行的程序代碼
2. 安全的代碼執(zhí)行環(huán)境：在隔離的受控環(huán)境中運(yùn)行生成的代碼
3. 執(zhí)行結(jié)果分析：獲取并解析代碼執(zhí)行的輸出與返回值
4. 迭代式優(yōu)化：基于執(zhí)行結(jié)果智能調(diào)整解決方案

打開網(wǎng)易新聞 查看精彩圖片

技術(shù)架構(gòu)與核心組件

CodeAct Agent的技術(shù)架構(gòu)由以下相互依存的關(guān)鍵組件構(gòu)成：

1. 代碼執(zhí)行環(huán)境：提供安全隔離的代碼運(yùn)行時(shí)環(huán)境，確保代碼執(zhí)行不影響宿主系統(tǒng)
2. 工作流定義系統(tǒng)：規(guī)范代碼生成、執(zhí)行與結(jié)果處理的邏輯流程
3. 提示工程機(jī)制：引導(dǎo)大語言模型以特定格式生成符合語法與邏輯要求的可執(zhí)行代碼
4. 狀態(tài)管理系統(tǒng)：維護(hù)對(duì)話歷史、執(zhí)行上下文與計(jì)算結(jié)果的持久化存儲(chǔ)

打開網(wǎng)易新聞 查看精彩圖片

代碼實(shí)現(xiàn)

1、基礎(chǔ)工具函數(shù)實(shí)現(xiàn)

首先需要定義代理可調(diào)用的基礎(chǔ)函數(shù)集。以下實(shí)現(xiàn)了一組基本數(shù)學(xué)運(yùn)算工具函數(shù)：

def add(a: int, b: int) -> int:
"""將兩個(gè)數(shù)字相加"""
return a + b
def subtract(a: int, b: int) -> int:
"""兩個(gè)數(shù)字相減"""
return a - b
def multiply(a: int, b: int) -> int:
"""兩個(gè)數(shù)字相乘"""
return a * b
def divide(a: int, b: int) -> float:
"""兩個(gè)數(shù)字相除"""
return a / b

2、代碼執(zhí)行環(huán)境構(gòu)建

接下來，實(shí)現(xiàn)代碼執(zhí)行環(huán)境，該環(huán)境需具備在多次調(diào)用間保持狀態(tài)的能力。SimpleCodeExecutor類實(shí)現(xiàn)了這一功能：

from typing import Any, Dict, Tuple
import io
import contextlib
import ast
import traceback
class SimpleCodeExecutor:
"""
一個(gè)簡單的代碼執(zhí)行器，可以在狀態(tài)def __init__(self, locals: Dict[str, Any], globals: Dict[str, Any]):
"""
初始化代碼執(zhí)行器。
參數(shù)：
locals: 在執(zhí)行上下文中使用的局部變量
globals: 在執(zhí)行上下文中使用的全局變量
"""
# 在執(zhí)行之間持久化的狀態(tài)
self.globals = globals
self.locals = locals
def execute(self, code: str) -> str:
"""
執(zhí)行Python代碼并捕獲輸出和返回值。
參數(shù)：
code: 要執(zhí)行的Python代碼
返回：
包含輸出和返回值的字符串
"""
# 捕獲標(biāo)準(zhǔn)輸出和標(biāo)準(zhǔn)錯(cuò)誤
stdout = io.StringIO()
stderr = io.StringIO()
output = ""
return_value = None
try:
# 執(zhí)行并捕獲輸出
with contextlib.redirect_stdout(
stdout
), contextlib.redirect_stderr(stderr):
# 嘗試檢測是否有返回值（最后一個(gè)表達(dá)式）
try:
tree = ast.parse(code)
last_node = tree.body[-1] if tree.body else None
# 如果最后一個(gè)語句是表達(dá)式，捕獲它的值
if isinstance(last_node, ast.Expr):
# 分割代碼以添加返回值賦值
last_line = code.rstrip().split("\n")[-1]
exec_code = (
code[: -len(last_line)]
+ "\n__result__ = "
+ last_line
)
# 執(zhí)行修改后的代碼
exec(exec_code, self.globals, self.locals)
return_value = self.locals.get("__result__")
else:
# 正常執(zhí)行
exec(code, self.globals, self.locals)
except:
# 如果解析失敗，按原樣執(zhí)行代碼
exec(code, self.globals, self.locals)
# 獲取輸出
output = stdout.getvalue()
if stderr.getvalue():
output += "\n" + stderr.getvalue()
except Exception as e:
# 捕獲異常信息
output = f"Error: {type(e).__name__}: {str(e)}\n"
output += traceback.format_exc()
if return_value is not None:
output += "\n\n" + str(return_value)
return output

3、工作流事件定義

為了規(guī)范化代碼執(zhí)行流程，需要定義控制工作流程的事件類型：

from llama_index.core.llms import ChatMessage
from llama_index.core.workflow import Event
class InputEvent(Event):
input: list[ChatMessage]
class StreamEvent(Event):
delta: str
class CodeExecutionEvent(Event):
code: str

4、CodeAct Agent工作流實(shí)現(xiàn)

下面是完整的CodeAct Agent工作流實(shí)現(xiàn)，它將所有組件整合為一個(gè)功能完整的系統(tǒng)：

import inspect
import re
from typing import Any, Callable, List
from llama_index.core.llms import ChatMessage, LLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import (
Context,
Workflow,
StartEvent,
StopEvent,
step,
)
from llama_index.llms.openai import OpenAI
CODEACT_SYSTEM_PROMPT = """
你是一個(gè)可以執(zhí)行代碼的有用助手。
根據(jù)聊天歷史，你可以在標(biāo)簽內(nèi)編寫代碼來幫助用戶解決問題。
在代碼中，你可以引用之前使用過的任何變量或函數(shù)。
用戶還為你提供了一些預(yù)定義函數(shù)：
{fn_str}
要執(zhí)行代碼，請(qǐng)?jiān)跇?biāo)簽之間編寫代碼。
"""
class CodeActAgent(Workflow):
def __init__(
self,
fns: List[Callable],
code_execute_fn: Callable,
llm: LLM | None = None,
**workflow_kwargs: Any,
) -> None:
super().__init__(**workflow_kwargs)
self.fns = fns or []
self.code_execute_fn = code_execute_fn
self.llm = llm or OpenAI(model="gpt-4o-mini")
# 將函數(shù)解析為截?cái)嗟暮瘮?shù)字符串
self.fn_str = "\n\n".join(
f'def {fn.__name__}{str(inspect.signature(fn))}:\n """ {fn.__doc__} """\n ...'
for fn in self.fns
)
self.system_message = ChatMessage(
role="system",
content=CODEACT_SYSTEM_PROMPT.format(fn_str=self.fn_str),
)
def _parse_code(self, response: str) -> str | None:
# 查找標(biāo)簽之間的代碼
matches = re.findall(r"", response, re.DOTALL)
if matches:
return "\n\n".join(matches)
return None
@step
async def prepare_chat_history(
self, ctx: Context, ev: StartEvent
) -> InputEvent:
# 檢查是否設(shè)置了內(nèi)存
memory = await ctx.get("memory", default=None)
if not memory:
memory = ChatMemoryBuffer.from_defaults(llm=self.llm)
# 獲取用戶輸入
user_input = ev.get("user_input")
if user_input is None:
raise ValueError("user_input kwarg is required")
user_msg = ChatMessage(role="user", content=user_input)
memory.put(user_msg)
# 獲取聊天歷史
chat_history = memory.get()
# 更新上下文
await ctx.set("memory", memory)
# 將系統(tǒng)消息添加到聊天歷史并返回
return InputEvent(input=[self.system_message, *chat_history])
@step
async def handle_llm_input(
self, ctx: Context, ev: InputEvent
) -> CodeExecutionEvent | StopEvent:
chat_history = ev.input
# 流式傳輸響應(yīng)
response_stream = await self.llm.astream_chat(chat_history)
async for response in response_stream:
ctx.write_event_to_stream(StreamEvent(delta=response.delta or ""))
# 保存最終響應(yīng)，應(yīng)包含所有內(nèi)容
memory = await ctx.get("memory")
memory.put(response.message)
await ctx.set("memory", memory)
# 獲取要執(zhí)行的代碼
code = self._parse_code(response.message.content)
if not code:
return StopEvent(result=response)
else:
return CodeExecutionEvent(code=code)
@step
async def handle_code_execution(
self, ctx: Context, ev: CodeExecutionEvent
) -> InputEvent:
# 執(zhí)行代碼
ctx.write_event_to_stream(ev)
output = self.code_execute_fn(ev.code)
# 更新內(nèi)存
memory = await ctx.get("memory")
memory.put(ChatMessage(role="assistant", content=output))
await ctx.set("memory", memory)
# 獲取最新的聊天歷史并循環(huán)回起點(diǎn)
chat_history = memory.get()
return InputEvent(input=[self.system_message, *chat_history])

5、CodeAct Agent的實(shí)例化與調(diào)用

完成組件構(gòu)建后，可以初始化并調(diào)用CodeAct Agent：

# 使用我們的函數(shù)初始化代碼執(zhí)行器
code_executor = SimpleCodeExecutor(
# 提供訪問我們的輔助函數(shù)
locals={
"add": add,
"subtract": subtract,
"multiply": multiply,
"divide": divide,
},
globals={
# 提供訪問所有內(nèi)置函數(shù)
"__builtins__": __builtins__,
# 提供訪問numpy
"np": __import__("numpy"),
},
)
# 創(chuàng)建代理
agent = CodeActAgent(
fns=[add, subtract, multiply, divide],
code_execute_fn=code_executor.execute,
llm=OpenAI(model="gpt-4o-mini", api_key="your_api_key_here"),
)
# 為代理創(chuàng)建上下文
ctx = Context(agent)
# 幫助函數(shù)，用于運(yùn)行代理并輸出詳細(xì)信息
async def run_agent_verbose(agent: CodeActAgent, ctx: Context, query: str):
handler = agent.run(user_input=query, ctx=ctx)
print(f"User: {query}")
async for event in handler.stream_events():
if isinstance(event, StreamEvent):
print(f"{event.delta}", end="", flush=True)
elif isinstance(event, CodeExecutionEvent):
print(f"\n-----------\nParsed code:\n{event.code}\n")
return await handler
# 運(yùn)行代理與示例
queries = [
"Calculate the sum of all numbers from 1 to 10",
"Add 5 and 3, then multiply the result by 2"
]
for query in queries:
response = await run_agent_verbose(agent, ctx, query)
print("\n" + "="*50 + "\n")

工作原理分析

CodeAct Agent的工作流程可分為以下關(guān)鍵階段：

1. 用戶輸入處理：系統(tǒng)接收用戶查詢，將其添加到對(duì)話記憶中，并結(jié)合系統(tǒng)提示準(zhǔn)備完整的對(duì)話上下文。
2. LLM代碼生成：大語言模型根據(jù)對(duì)話上下文生成可能包含代碼塊的響應(yīng)。系統(tǒng)通過特定標(biāo)簽精確識(shí)別可執(zhí)行代碼段。若未檢測到代碼塊，工作流程終止并返回純文本響應(yīng)。
4. 代碼執(zhí)行階段：系統(tǒng)將識(shí)別的代碼發(fā)送至執(zhí)行器，在受控環(huán)境中運(yùn)行代碼，捕獲執(zhí)行輸出和錯(cuò)誤信息，并將結(jié)果保存至對(duì)話記憶。
5. 響應(yīng)迭代處理：工作流程循環(huán)返回LLM處理階段，使模型能夠訪問代碼執(zhí)行結(jié)果，并根據(jù)需要繼續(xù)生成響應(yīng)。此迭代過程持續(xù)進(jìn)行，直至生成不包含代碼的最終響應(yīng)。

安全性考量

本文展示的SimpleCodeExecutor僅適用于開發(fā)環(huán)境。生產(chǎn)環(huán)境部署時(shí)應(yīng)考慮以下安全措施：

1. 實(shí)現(xiàn)容器化隔離環(huán)境（如Docker），確保代碼執(zhí)行不影響宿主系統(tǒng)
2. 配置嚴(yán)格的資源限制機(jī)制，包括CPU使用率、內(nèi)存上限及最大執(zhí)行時(shí)間
3. 實(shí)施精細(xì)的權(quán)限控制，嚴(yán)格限制對(duì)敏感模塊和系統(tǒng)函數(shù)的訪問
4. 構(gòu)建輸入驗(yàn)證和安全過濾系統(tǒng)，防止注入攻擊

結(jié)語

本文詳細(xì)闡述了基于LlamaIndex構(gòu)建CodeAct Agent的完整技術(shù)方案，從代碼執(zhí)行環(huán)境的構(gòu)建、工作流事件的定義到完整Agent的實(shí)現(xiàn)。通過將代碼生成與執(zhí)行能力無縫集成到對(duì)話式AI系統(tǒng)中，CodeAct Agent代表了一種新型交互范式，能夠?qū)⒆匀徽Z言指令轉(zhuǎn)化為可執(zhí)行的計(jì)算邏輯。

這一技術(shù)架構(gòu)的核心價(jià)值在于其可擴(kuò)展性和靈活性。開發(fā)者可以根據(jù)特定應(yīng)用場景定制執(zhí)行環(huán)境、函數(shù)庫和安全策略，從而構(gòu)建出專用的智能工具。隨著大語言模型能力的不斷提升，CodeAct Agent的應(yīng)用前景將更加廣闊，特別是在以下領(lǐng)域：

• 專業(yè)領(lǐng)域軟件開發(fā)輔助工具
• 數(shù)據(jù)科學(xué)探索與可視化系統(tǒng)
• 教育領(lǐng)域的編程學(xué)習(xí)平臺(tái)
• 企業(yè)級(jí)自動(dòng)化工作流構(gòu)建

未來CodeAct技術(shù)的發(fā)展方向包括更精細(xì)的代碼生成控制、多語言執(zhí)行環(huán)境支持、更強(qiáng)大的安全隔離機(jī)制以及與專業(yè)領(lǐng)域知識(shí)庫的深度集成。這些進(jìn)步將進(jìn)一步拓展AI輔助編程的邊界，使自然語言與代碼執(zhí)行之間的轉(zhuǎn)換更加高效、安全和可靠。

通過深入理解CodeAct Agent的核心技術(shù)架構(gòu)，開發(fā)者能夠構(gòu)建將大語言模型的自然語言理解能力與代碼執(zhí)行功能有機(jī)結(jié)合的高度定制化解決方案，為各類應(yīng)用場景提供強(qiáng)大的智能化支持。

https://avoid.overfit.cn/post/d66bc1e556934aa1b32a3ca58983795c