感觉阿里的AgentScope和LangGraph的设计，还是有一类似，可以好好学学。这次先按官方文档进行实操。

一，模型配置

from dotenv import load_dotenv
import asyncio
import os
from agentscope.model import DeepSeekChatModel
from agentscope.credential import DeepSeekCredential
from agentscope.formatter import OpenAIMultiAgentFormatter
from agentscope.message import UserMsg
from agentscope.model import FinishedReason
from pydantic import BaseModel

load_dotenv(verbose=True)
api_key=os.getenv("DEEPSEEK_API_KEY")
model = DeepSeekChatModel(
    credential=DeepSeekCredential(api_key=api_key),
    model='deepseek-v4-flash',
    stream=True,
    formatter=OpenAIMultiAgentFormatter(),
    )
msgs = [UserMsg(name="user", content="Count from 1 to 5.")]

class WeatherInfo(BaseModel):
    city: str
    temperature: float
    units: str

async def main():
    '''
    response = await model(msgs)
    print(response.content)   # [TextBlock(text='1, 2, 3, 4, 5')]

    async  for chunk in await model(msgs):
        if chunk.is_last:
            print('Final:', chunk.content)
        else:
            print('Delta:', chunk.content)
     '''
    response = await model.generate_structured_output(
        messages=[UserMsg(name="user", content="上海今天天气怎么样？")],
        structured_model=WeatherInfo,
    )
    print(response)

async def call_model():
    async for chunk in await model(msgs):
        if chunk.is_last and chunk.finished_reason == FinishedReason.INTERRUPTED:
            # 取消前已累积的部分内容
            print('Interrupted:', chunk.content)


# task = asyncio.create_task(call_model())
# 取消任务即可中断进行中的模型调用
# task.cancel()
asyncio.run(main())

二，上下文

from dotenv import load_dotenv
import asyncio
import os
from agentscope.agent import Agent, InjectionConfig, ContextConfig
from agentscope.model import DeepSeekChatModel
from agentscope.credential import DeepSeekCredential
from agentscope.tool import Toolkit, Bash, Read, Write, Edit
from agentscope.workspace import LocalWorkspace

load_dotenv(verbose=True)
api_key=os.getenv("DEEPSEEK_API_KEY")

workspace=LocalWorkspace(workdir='D:\\testcode\\')
# await workspace.initialize()

model = DeepSeekChatModel(
    credential=DeepSeekCredential(api_key=api_key),
    model='deepseek-v4-flash',
    stream=True,
    )
toolkit = Toolkit(tools=[Bash(), Read(), Write(), Edit()])

agent = Agent(
    name='my_agent',
    system_prompt='你是一个有帮助的助手',
    model=model,
    context_config=ContextConfig(
        trigger_ratio=0.8,
        reserve_ratio=0.1,
        tool_result_limit=3000,
    ),
    toolkit=toolkit,
    injection_config=InjectionConfig(
        timezone='Asia/Shanghai',
        time_interval=1.0,
    ),
    offloader=workspace
)

三， 函数工具

from typing import AsyncGenerator, Any, Callable

from redis.commands.search import result

from agentscope.tool import FunctionTool, Toolkit, ToolMiddlewareBase, ToolBase, ToolChunk
from agentscope.tool import Bash, PowerShell
from agentscope.permission import (
    PermissionContext, PermissionDecision, PermissionBehavior
)
from agentscope.message import TextBlock


bash = Bash(
    additional_dangerous_files=['.secrets'],
    additional_dangerous_directories=[".credentials"],
)
pwsh = PowerShell(cmd='C:\\users\me\\project')

class WebSearch(ToolBase):
    name = 'WebSearch'
    description = 'Search the web for information on a given query.'
    input_schema = {
        'type': 'object',
        'properties': {
            'query': {
                'type': 'string',
                'description': 'The Search query.',
            },
        },
        'required': ['query'],
    }
    is_concurrency_safe = True
    is_read_only = True

    async def check_permissions(
        self,
        tool_input: dict[str, Any],
        context: PermissionContext,
    ) -> PermissionDecision:
        return PermissionDecision(
            behavior=PermissionBehavior.ALLOW,
            message='Web search is read-only.',
        )
    async  def call(self, query: str) -> ToolChunk:
        # results = await do_search(query) # 需要实现的异步操作
        results = 'demo'
        return ToolChunk(content=[TextBlock(text=results)])


def get_weather(city:str, unit:str = 'celsius') -> str:
    """
    Get weather data for given city and unit.
    Args:
        city (str): City name.
        unit (str, optional): Unit name. Defaults to 'celsius'.
    """
    return f'The weather for {city} is {unit} celsius.'
toolkit = Toolkit(tools=[FunctionTool(get_weather)])



class LoggingMiddleware(ToolMiddlewareBase):
    async def on_tool_call(self,
                         tool: ToolBase,
                         input_kwargs: dict[str, Any],
                         next_handler: Callable[..., AsyncGenerator[ToolChunk, None]],
                         ) -> AsyncGenerator[ToolChunk, None]:
        print(f'-> 调用{tool.name}，参数：{input_kwargs}')
        async for chunk in next_handler(**input_kwargs):
            yield chunk
        print(f'OK,{tool.name}执行完毕')

class RetryMiddleware(ToolMiddlewareBase):
    def __init__(self, max_attempts: int = 3):
        self.max_attempts = max_attempts

    async def on_tool_call(self,
                         tool: ToolBase,
                         input_kwargs: dict[str, Any],
                         next_handler: Callable[..., AsyncGenerator[ToolChunk, None]],
                         ) -> AsyncGenerator[ToolChunk, None]:
        for attempt in range(1, self.max_attempts + 1):
            try:
                async for chunk in next_handler(**input_kwargs):
                    yield chunk
                return
            except Exception as e:
                if attempt == self.max_attempts:
                    print(f'第{attempt}次失败：{e}，重试中...')
        print(f'-> 调用{tool.name}，参数：{input_kwargs}')

bash = Bash(middlewares=[LoggingMiddleware(), RetryMiddleware(max_attempts=3)])

from cryptography.x509 import name

from agentscope.tool import Toolkit, ToolGroup, Bash, Read, Write, Edit

db_query_tool = [Bash(),  Write(), Edit()],
db_migrate_tool = [Read(), Write(), Edit()],
deploy_tool = [Bash(), Read(),  Edit()],
rollback_tool = [Bash(), Read(), Write()],

toolkit = Toolkit(
    tool_groups=[
        ToolGroup(
            name='database',
            description='Tools for database operations',
            instructions='Always wrap mutations in a transaction',
            tools=[db_query_tool, db_migrate_tool],
        ),
    ToolGroup(
            name='deployment',
            description='Tools fordeploying services.',
            instructions='Confirm the target environment before deploying ',
            tools=[deploy_tool, rollback_tool],
        ),
    ]
)

四 ，MCP

from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.tool import Toolkit

client = MCPClient(
    name="search",
    is_stateful=False,
    mcp_config=HttpMCPConfig(url='https://api.search.com/mcp'),
    enable_tools=['web_search', 'image_search'],
)

toolkit = Toolkit(mcps=[client])

五，SKILL

from agentscope.tool import Toolkit
from agentscope.skill import LocalSkillLoader

skill_loader = LocalSkillLoader(
    directory='/path/to/skills',
    scan_subdir=True,
)

toolkit = Toolkit(skills_or_loaders=[skill_loader])

六 Plan

from agentscope.agent import Agent
from agentscope.state import  Task
from agentscope.tool import Toolkit, TaskGet, TaskCreate, TaskList, TaskUpdate

agent = Agent(
    name='planner',
    system_prompt='You are a planning assistant.',
    toolkit=Toolkit(
        tools=[TaskCreate(), TaskGet(), TaskList(), TaskUpdate()],
    )
)

agent.state.tasks_context.tasks.extend(
    [
        Task(
            id='1',
            subject='Fetch project requirements',
            description='Read README.md and CONTRIBUTIING.md in the repo root.',
            metadata={'source': 'seed'}
        ),
        Task(
            id='2',
            subject='Draft an inplementation plan',
            description='Produce a step-by-step plan based on the requirements.',
            metadata={'source': 'seed'}
        ),
    ]
)

agent.state.tasks_context.tasks[1].blocks.append('2')