Amazon Bedrock agentcore 代码解释器集成

import os

os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"

from langchain_aws.tools import create_code_interpreter_toolkit

# 创建工具包并获取工具（异步）
toolkit, code_tools = await create_code_interpreter_toolkit(region="us-west-2")

# 按名称获取工具
tools_by_name = toolkit.get_tools_by_name()

# 执行 Python 代码
result = tools_by_name["execute_code"].invoke({
    "code": """
import numpy as np
data = [1, 2, 3, 4, 5]
print(f"Mean: {np.mean(data)}")
print(f"Sum: {np.sum(data)}")
""",
    "language": "python"
})
print(result)

import asyncio
from langchain.agents import create_react_agent
from langchain.chat_models import init_chat_model
from langchain_aws.tools import create_code_interpreter_toolkit

async def main():
    # 创建工具包
    toolkit, code_tools = await create_code_interpreter_toolkit(region="us-west-2")

    # 初始化聊天模型
    llm = init_chat_model(
        "us.anthropic.claude-sonnet-4-20250514-v1:0",
        model_provider="bedrock_converse",
    )

    # 使用代码解释器工具创建智能体
    agent = create_react_agent(
        model=llm,
        tools=code_tools,
    )

    # 创建带有 thread_id 的配置以实现会话隔离
    config = {"configurable": {"thread_id": "session-123"}}

    # 运行智能体
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Calculate the factorial of 10"}]},
        config=config
    )
    print(result["messages"][-1].content)

    # 完成后清理
    await toolkit.cleanup()

asyncio.run(main())

# 不同的线程具有隔离的会话
config_user1 = {"configurable": {"thread_id": "user-1"}}
config_user2 = {"configurable": {"thread_id": "user-2"}}

# 在 user-1 会话中定义的变量在 user-2 会话中不存在
await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Set x = 100"}]},
    config=config_user1
)

await agent.ainvoke(
    {"messages": [{"role": "user", "content": "What is x?"}]},  # x 在此处未定义
    config=config_user2
)

tools_by_name = toolkit.get_tools_by_name()

# 写入文件
tools_by_name["write_files"].invoke({
    "files": [{"path": "data.csv", "text": "name,value\nAlice,100\nBob,200"}]
})

# 读取回来
content = tools_by_name["read_files"].invoke({"paths": ["data.csv"]})
print(content)

# 列出当前目录中的文件
files = tools_by_name["list_files"].invoke({"directory_path": "."})
print(files)

tools_by_name["upload_file"].invoke({
    "path": "sales.csv",
    "content": "date,revenue,product\n2024-01-01,1000,Widget\n2024-01-02,1500,Gadget",
    "description": "Sales data with columns: date, revenue, product_id"
})

tools_by_name["install_packages"].invoke({
    "packages": ["pandas>=2.0", "matplotlib", "scikit-learn"],
    "upgrade": False
})

tools_by_name = toolkit.get_tools_by_name()
config = {"configurable": {"thread_id": "session-123"}}

# 异步启动一个长时间运行的命令
result = tools_by_name["start_command_execution"].invoke(
    {"command": "python long_running_script.py"},
    config=config
)
# 返回一个 task_id

# 检查任务状态
status = tools_by_name["get_task"].invoke(
    {"task_id": "task-abc123"},
    config=config
)
print(status)

# 如果需要，停止正在运行的任务
tools_by_name["stop_task"].invoke(
    {"task_id": "task-abc123"},
    config=config
)

# 清理所有会话
await toolkit.cleanup()

# 或清理特定线程的会话
await toolkit.cleanup(thread_id="session-123")