Building Agents with Qwen

pip install transformers

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, Agent
from transformers.generation import GenerationConfig

class QWenAgent(Agent):
    """
    Agent that uses QWen model and tokenizer to generate code.
    
    Args:
        chat_prompt_template (str, optional): Custom chat prompt template
        run_prompt_template (str, optional): Custom run prompt template
        additional_tools (list, optional): Additional tools beyond defaults
    
    Example:
        agent = QWenAgent()
        agent.run("Draw me a picture of rivers and lakes.")
    """
    def __init__(self, chat_prompt_template=None, run_prompt_template=None, additional_tools=None):
        checkpoint = "Qwen/Qwen-7B-Chat"
        self.tokenizer = AutoTokenizer.from_pretrained(
            checkpoint,
            trust_remote_code=True
        )
        self.model = AutoModelForCausalLM.from_pretrained(
            checkpoint,
            device_map="auto",
            trust_remote_code=True
        ).cuda().eval()
        
        self.model.generation_config = GenerationConfig.from_pretrained(
            checkpoint,
            trust_remote_code=True
        )
        self.model.generation_config.do_sample = False  # Use greedy decoding
        
        super().__init__(
            chat_prompt_template=chat_prompt_template,
            run_prompt_template=run_prompt_template,
            additional_tools=additional_tools,
        )
    
    def generate_one(self, prompt, stop):
        # Replace special tokens (legacy requirement, will be fixed)
        prompt = prompt.replace("Human:", "_HUMAN_:").replace("Assistant:", "_ASSISTANT_:")
        stop = [item.replace("Human:", "_HUMAN_:").replace("Assistant:", "_ASSISTANT_:") for item in stop]
        
        result, _ = self.model.chat(self.tokenizer, prompt, history=None)
        
        # Remove stop sequences
        for stop_seq in stop:
            if result.endswith(stop_seq):
                result = result[:-len(stop_seq)]
        
        # Restore special tokens
        result = result.replace("_HUMAN_:", "Human:").replace("_ASSISTANT_:", "Assistant:")
        return result

# Remote execution (HuggingFace API)
agent.run("transcribe this audio", remote=True)

# Local execution (downloads and runs model locally)
agent.run("transcribe this audio", remote=False)

import json
from transformers import AutoModelForCausalLM, AutoTokenizer

def build_react_agent(tools):
    """
    Create a ReAct agent with specified tools.
    
    Args:
        tools: List of tool definitions
    
    Returns:
        Agent function that processes queries
    """
    model_name = "Qwen/Qwen-7B-Chat"
    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        device_map="auto",
        trust_remote_code=True
    ).eval()
    
    def agent(query, history=None):
        if history is None:
            history = []
        
        # Build ReAct prompt with tools
        prompt = build_react_prompt(query, tools, history)
        
        # Generate with stop at Observation
        response, _ = model.chat(
            tokenizer,
            prompt,
            history=None,
            stop_words_ids=[
                tokenizer.encode('Observation:'),
                tokenizer.encode('Observation:\n')
            ]
        )
        
        # Parse action and execute tool
        action, args = parse_action(response)
        if action:
            observation = execute_tool(action, args, tools)
            # Continue generation with observation
            final_response, _ = model.chat(
                tokenizer,
                prompt + response + f"\nObservation: {observation}\nThought:",
                history=None
            )
            return final_response
        
        return response
    
    return agent

# Create and use agent
tools = [
    {
        'name': 'search',
        'description': 'Search the internet for information',
        'parameters': {'query': 'string'}
    }
]

agent = build_react_agent(tools)
response = agent("What is the capital of France?")

from langchain.agents import AgentType, initialize_agent, load_tools
from langchain.chat_models import ChatOpenAI

# Configure Qwen as LangChain LLM
llm = ChatOpenAI(
    model_name='Qwen',
    openai_api_base='http://localhost:8000/v1',
    openai_api_key='EMPTY',
    streaming=False,
)

# Load tools
tools = load_tools(['arxiv', 'wikipedia'])

# Create agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    max_iterations=3,
    early_stopping_method="generate"
)

# Run agent
result = agent.run('查一下论文 1605.08386 的信息')
print(result)

class TaskDecompositionAgent:
    """Agent that breaks complex tasks into subtasks."""
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
    
    def decompose_task(self, task: str) -> list:
        """Break task into subtasks."""
        prompt = f"""Break down this task into smaller steps:
        Task: {task}
        
        Steps:
        1."""
        
        response, _ = self.model.chat(self.tokenizer, prompt, history=None)
        return self.parse_steps(response)
    
    def execute_subtask(self, subtask: str, context: dict) -> str:
        """Execute a single subtask."""
        prompt = f"""Complete this subtask:
        Subtask: {subtask}
        Context: {json.dumps(context)}
        
        Result:"""
        
        result, _ = self.model.chat(self.tokenizer, prompt, history=None)
        return result
    
    def run(self, task: str) -> str:
        """Execute complete task."""
        subtasks = self.decompose_task(task)
        context = {}
        
        for subtask in subtasks:
            result = self.execute_subtask(subtask, context)
            context[subtask] = result
        
        return self.synthesize_results(context)

class MemoryAgent:
    """Agent with episodic memory."""
    
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.memory = []
    
    def remember(self, interaction: dict):
        """Store interaction in memory."""
        self.memory.append(interaction)
        if len(self.memory) > 10:  # Keep last 10
            self.memory.pop(0)
    
    def recall(self, query: str) -> list:
        """Retrieve relevant memories."""
        # Simple keyword matching (can be enhanced with embeddings)
        relevant = []
        for mem in self.memory:
            if any(word in mem['query'].lower() for word in query.lower().split()):
                relevant.append(mem)
        return relevant
    
    def chat(self, query: str) -> str:
        """Chat with memory context."""
        # Retrieve relevant memories
        memories = self.recall(query)
        
        # Build context-aware prompt
        context = "\n".join([
            f"Previous: Q: {m['query']} A: {m['response']}"
            for m in memories
        ])
        
        if context:
            prompt = f"{context}\n\nCurrent question: {query}"
        else:
            prompt = query
        
        # Generate response
        response, _ = self.model.chat(self.tokenizer, prompt, history=None)
        
        # Store interaction
        self.remember({'query': query, 'response': response})
        
        return response

from transformers import AutoModelForCausalLM, AutoTokenizer

# Initialize model
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-7B-Chat", trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen-7B-Chat",
    device_map="auto",
    trust_remote_code=True
).eval()

# Define tools
tools = [
    {
        'name_for_human': '谷歌搜索',
        'name_for_model': 'google_search',
        'description_for_model': '谷歌搜索是一个通用搜索引擎，可用于访问互联网、查询百科知识、了解时事新闻等。',
        'parameters': [{'name': 'search_query', 'description': '搜索关键词', 'required': True, 'schema': {'type': 'string'}}],
    },
    {
        'name_for_human': '文生图',
        'name_for_model': 'image_gen',
        'description_for_model': '文生图是一个AI绘画服务，输入文本描述，返回图片URL',
        'parameters': [{'name': 'prompt', 'description': '英文关键词', 'required': True, 'schema': {'type': 'string'}}],
    },
]

# Run agent
history = []
for query in ['你好', '搜索一下谁是周杰伦', '给我画个可爱的小猫']:
    response, history = llm_with_plugin(
        prompt=query,
        history=history,
        list_of_plugin_info=tools
    )
    print(f"User: {query}")
    print(f"Agent: {response}\n")